1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
|
import shutil
import tempfile
import unittest
import openpaperwork_core
class TestPageTracker(unittest.TestCase):
def setUp(self):
self.tmp_paperwork_dir = tempfile.mkdtemp(
prefix="paperwork_backend_tests"
)
self.core = openpaperwork_core.Core(auto_load_dependencies=True)
class FakeModule(object):
class Plugin(openpaperwork_core.PluginBase):
PRIORITY = 999999999999999999999
def data_dir_handler_get_individual_data_dir(s):
return openpaperwork_core.fs.CommonFsPluginBase.fs_safe(
self.tmp_paperwork_dir
)
self.core._load_module("fake_module", FakeModule)
self.core.load("paperwork_backend.model.fake")
self.core.load("paperwork_backend.pagetracker")
self.fake_storage = self.core.get_by_name(
"paperwork_backend.model.fake"
)
self.core.init()
def tearDown(self):
shutil.rmtree(self.tmp_paperwork_dir)
def test_tracking(self):
self.fake_storage.docs = [
{
'id': 'test_doc',
'url': 'file:///somewhere/test_doc',
'page_hashes': [
('file:///somewhere/test_doc/0.jpeg', 123),
('file:///somewhere/test_doc/1.jpeg', 124),
],
},
{
'id': 'test_doc_2',
'url': 'file:///somewhere/test_doc_2',
'page_hashes': [
('file:///somewhere/test_doc_2/0.jpeg', 125),
('file:///somewhere/test_doc_2/1.jpeg', 126),
('file:///somewhere/test_doc_2/2.jpeg', 127),
],
},
]
tracker = self.core.call_success("page_tracker_get", 'test_tracking')
out = tracker.find_changes('test_doc', 'file:///somewhere/test_doc')
self.assertEqual(out, [('new', 0), ('new', 1)])
tracker.ack_page('test_doc', 'file:///somewhere/test_doc', 0)
tracker.ack_page('test_doc', 'file:///somewhere/test_doc', 1)
out = tracker.find_changes(
'test_doc_2', 'file:///somewhere/test_doc_2'
)
self.assertEqual(out, [('new', 0), ('new', 1), ('new', 2)])
tracker.ack_page('test_doc_2', 'file:///somewhere/test_doc_2', 0)
tracker.ack_page('test_doc_2', 'file:///somewhere/test_doc_2', 1)
tracker.ack_page('test_doc_2', 'file:///somewhere/test_doc_2', 2)
tracker.commit()
self.fake_storage.docs = [
{
'id': 'test_doc',
'url': 'file:///somewhere/test_doc',
'page_hashes': [
('file:///somewhere/test_doc/0.jpeg', 256),
('file:///somewhere/test_doc/1.jpeg', 124),
('file:///somewhere/test_doc/2.jpeg', 257),
],
},
]
tracker = self.core.call_success("page_tracker_get", 'test_tracking')
out = tracker.find_changes('test_doc', 'file:///somewhere/test_doc')
self.assertEqual(out, [('upd', 0), ('new', 2)])
tracker.ack_page('test_doc', 'file:///somewhere/test_doc', 0)
tracker.ack_page('test_doc', 'file:///somewhere/test_doc', 2)
tracker.delete_doc('test_doc_2')
tracker.commit()
|
