1import hashlib
2import json
3from pathlib import Path
4import tempfile
5import unittest
6import xml.etree.ElementTree as ET
7
8from native_xml import ns
9from verify_smb_faults import verify
10
11
12class NativeFaultOracle(unittest.TestCase):
13 def setUp(self):
14 temporary = tempfile.TemporaryDirectory()
15 self.addCleanup(temporary.cleanup)
16 self.root = Path(temporary.name)
17 self.cases = self.root / 'cases'
18 self.worker = self.root / 'captures/worker-0'
19 self.worker.mkdir(parents=True)
20 self.records = [{'case': 'sample-000', 'visible': 'before'},
21 {'case': 'sample-001', 'visible': 'after'}]
22 self.results = []
23 for role, names in [('source', ['sample']), ('interrupted', ['sample-000', 'sample-001']),
24 ('recovered', ['sample-000', 'sample-001'])]:
25 (self.cases / role).mkdir(parents=True)
26 for name in names:
27 content = (role + name).encode()
28 (self.cases / role / (name + '.one')).write_bytes(content)
29 name = role + '-' + name
30 self.results.append({'name': name, 'error': None, 'pages': 1,
31 'source_sha256': hashlib.sha256(content).hexdigest()})
32 replacement = 'before'
33 if name == 'interrupted-sample-001': replacement = 'after'
34 if role == 'recovered': replacement = 'after again'
35 self.page(name, ['Title', replacement, 'Unrelated café 🦀'])
36 (self.cases / 'sample-intent.json').write_text(json.dumps({'before': 'before', 'after': 'after', 'suffix': ' again'}))
37 self.save()
38
39 def save(self):
40 (self.cases / 'results.json').write_text(json.dumps(self.records))
41 (self.worker / 'results.json').write_text(json.dumps(self.results))
42
43 def page(self, name, paragraphs):
44 page = ET.Element('{' + ns['one'] + '}Page')
45 outline = ET.SubElement(page, '{' + ns['one'] + '}Outline')
46 for text in paragraphs:
47 ET.SubElement(outline, '{' + ns['one'] + '}T').text = text
48 output = self.worker / 'results' / name
49 output.mkdir(parents=True, exist_ok=True)
50 ET.ElementTree(page).write(output / 'page-0.xml', encoding='utf-8')
51
52 def check(self):
53 return verify(self.root, self.root / 'captures')
54
55 def test_complete_before_and_after_outcomes(self):
56 self.assertEqual(self.check(), {'cases': 2, 'cold_native_opens': 5, 'exact_native_text': True})
57
58 def test_unrelated_or_partial_edits_are_rejected(self):
59 for paragraphs in [['Title', 'after again', 'Changed'], ['Title', 'afte', 'Unrelated café 🦀'],
60 ['Title', 'after again again', 'Unrelated café 🦀']]:
61 with self.subTest(paragraphs=paragraphs):
62 self.page('recovered-sample-000', paragraphs)
63 with self.assertRaises(AssertionError): self.check()
64
65 def test_wrong_publication_outcome_is_rejected(self):
66 self.records[0]['visible'] = 'after'
67 self.save()
68 with self.assertRaises(AssertionError): self.check()
69
70 def test_input_hash_is_checked(self):
71 (self.cases / 'interrupted/sample-000.one').write_bytes(b'changed')
72 with self.assertRaises(AssertionError): self.check()
73
74 def test_missing_and_duplicate_captures_are_rejected(self):
75 original = self.results[:]
76 for rows in [original[:-1], original + original[:1]]:
77 self.results = rows
78 self.save()
79 with self.assertRaises(AssertionError): self.check()
80
81 def test_extra_artifacts_are_rejected(self):
82 (self.cases / 'interrupted/extra.one').write_bytes(b'extra')
83 with self.assertRaises(AssertionError): self.check()
84
85 def test_duplicate_cases_are_rejected(self):
86 self.records.append(self.records[0])
87 self.save()
88 with self.assertRaises(AssertionError): self.check()
89
90 def test_native_failure_and_wrong_page_count_are_rejected(self):
91 for error, pages in [('Open failed', 1), (None, 0), (None, 2)]:
92 self.results[0].update(error=error, pages=pages)
93 self.save()
94 with self.assertRaises(AssertionError): self.check()
95
96
97if __name__ == '__main__': unittest.main()