9 import notmuch2._errors as errors
10 import notmuch2._database as dbmod
11 import notmuch2._message as message
16 with dbmod.Database.create(maildir.path) as db:
21 """Tests for reading the default database.
23 The error cases are fairly undefined, some relevant Python error
24 will come out if you give it a bad filename or if the file does
25 not parse correctly. So we're not testing this too deeply.
28 def test_config_pathname_default(self, monkeypatch):
29 monkeypatch.delenv('NOTMUCH_CONFIG', raising=False)
30 user = pathlib.Path('~/.notmuch-config').expanduser()
31 assert dbmod._config_pathname() == user
33 def test_config_pathname_env(self, monkeypatch):
34 monkeypatch.setenv('NOTMUCH_CONFIG', '/some/random/path')
35 assert dbmod._config_pathname() == pathlib.Path('/some/random/path')
37 def test_default_path_nocfg(self, monkeypatch, tmppath):
38 monkeypatch.setenv('NOTMUCH_CONFIG', str(tmppath/'foo'))
39 with pytest.raises(FileNotFoundError):
40 dbmod.Database.default_path()
42 def test_default_path_cfg_is_dir(self, monkeypatch, tmppath):
43 monkeypatch.setenv('NOTMUCH_CONFIG', str(tmppath))
44 with pytest.raises(IsADirectoryError):
45 dbmod.Database.default_path()
47 def test_default_path_parseerr(self, monkeypatch, tmppath):
48 cfg = tmppath / 'notmuch-config'
49 with cfg.open('w') as fp:
51 monkeypatch.setenv('NOTMUCH_CONFIG', str(cfg))
52 with pytest.raises(configparser.Error):
53 dbmod.Database.default_path()
55 def test_default_path_parse(self, monkeypatch, tmppath):
56 cfg = tmppath / 'notmuch-config'
57 with cfg.open('w') as fp:
58 fp.write('[database]\n')
59 fp.write('path={!s}'.format(tmppath))
60 monkeypatch.setenv('NOTMUCH_CONFIG', str(cfg))
61 assert dbmod.Database.default_path() == tmppath
63 def test_default_path_param(self, monkeypatch, tmppath):
64 cfg_dummy = tmppath / 'dummy'
65 monkeypatch.setenv('NOTMUCH_CONFIG', str(cfg_dummy))
66 cfg_real = tmppath / 'notmuch_config'
67 with cfg_real.open('w') as fp:
68 fp.write('[database]\n')
69 fp.write('path={!s}'.format(cfg_real/'mail'))
70 assert dbmod.Database.default_path(cfg_real) == cfg_real/'mail'
75 def test_create(self, tmppath, db):
76 assert tmppath.joinpath('.notmuch/xapian/').exists()
78 def test_create_already_open(self, tmppath, db):
79 with pytest.raises(errors.NotmuchError):
82 def test_create_existing(self, tmppath, db):
83 with pytest.raises(errors.FileError):
84 dbmod.Database.create(path=tmppath)
86 def test_close(self, db):
89 def test_del_noclose(self, db):
92 def test_close_del(self, db):
96 def test_closed_attr(self, db):
101 def test_ctx(self, db):
107 def test_path(self, db, tmppath):
108 assert db.path == tmppath
110 def test_version(self, db):
111 assert db.version > 0
113 def test_needs_upgrade(self, db):
114 assert db.needs_upgrade in (True, False)
119 def test_exit_early(self, db):
120 with pytest.raises(errors.UnbalancedAtomicError):
121 with db.atomic() as ctx:
124 def test_exit_late(self, db):
125 with db.atomic() as ctx:
127 with pytest.raises(errors.UnbalancedAtomicError):
133 def test_single_rev(self, db):
135 assert isinstance(r, dbmod.DbRevision)
136 assert isinstance(r.rev, int)
137 assert isinstance(r.uuid, bytes)
145 def test_diff_db(self, tmppath):
146 dbpath0 = tmppath.joinpath('db0')
148 dbpath1 = tmppath.joinpath('db1')
150 db0 = dbmod.Database.create(path=dbpath0)
151 db1 = dbmod.Database.create(path=dbpath1)
152 r_db0 = db0.revision()
153 r_db1 = db1.revision()
154 assert r_db0 != r_db1
155 assert r_db0.uuid != r_db1.uuid
157 def test_cmp(self, db, maildir):
159 _, pathname = maildir.deliver()
160 db.add(pathname, sync_flags=False)
164 assert not rev0 > rev1
165 assert not rev0 >= rev1
166 assert not rev0 == rev1
169 # XXX add tests for revisions comparisons
173 def test_add_message(self, db, maildir):
174 msgid, pathname = maildir.deliver()
175 msg, dup = db.add(pathname, sync_flags=False)
176 assert isinstance(msg, message.Message)
177 assert msg.path == pathname
178 assert msg.messageid == msgid
180 def test_add_message_str(self, db, maildir):
181 msgid, pathname = maildir.deliver()
182 msg, dup = db.add(str(pathname), sync_flags=False)
184 def test_add_message_bytes(self, db, maildir):
185 msgid, pathname = maildir.deliver()
186 msg, dup = db.add(os.fsencode(bytes(pathname)), sync_flags=False)
188 def test_remove_message(self, db, maildir):
189 msgid, pathname = maildir.deliver()
190 msg, dup = db.add(pathname, sync_flags=False)
191 assert db.find(msgid)
192 dup = db.remove(pathname)
193 with pytest.raises(LookupError):
196 def test_remove_message_str(self, db, maildir):
197 msgid, pathname = maildir.deliver()
198 msg, dup = db.add(pathname, sync_flags=False)
199 assert db.find(msgid)
200 dup = db.remove(str(pathname))
201 with pytest.raises(LookupError):
204 def test_remove_message_bytes(self, db, maildir):
205 msgid, pathname = maildir.deliver()
206 msg, dup = db.add(pathname, sync_flags=False)
207 assert db.find(msgid)
208 dup = db.remove(os.fsencode(bytes(pathname)))
209 with pytest.raises(LookupError):
212 def test_find_message(self, db, maildir):
213 msgid, pathname = maildir.deliver()
214 msg0, dup = db.add(pathname, sync_flags=False)
215 msg1 = db.find(msgid)
216 assert isinstance(msg1, message.Message)
217 assert msg1.messageid == msgid == msg0.messageid
218 assert msg1.path == pathname == msg0.path
220 def test_find_message_notfound(self, db):
221 with pytest.raises(LookupError):
224 def test_get_message(self, db, maildir):
225 msgid, pathname = maildir.deliver()
226 msg0, _ = db.add(pathname, sync_flags=False)
227 msg1 = db.get(pathname)
228 assert isinstance(msg1, message.Message)
229 assert msg1.messageid == msgid == msg0.messageid
230 assert msg1.path == pathname == msg0.path
232 def test_get_message_str(self, db, maildir):
233 msgid, pathname = maildir.deliver()
234 db.add(pathname, sync_flags=False)
235 msg = db.get(str(pathname))
236 assert msg.messageid == msgid
238 def test_get_message_bytes(self, db, maildir):
239 msgid, pathname = maildir.deliver()
240 db.add(pathname, sync_flags=False)
241 msg = db.get(os.fsencode(bytes(pathname)))
242 assert msg.messageid == msgid
246 # We just want to test this behaves like a set at a hight level.
247 # The set semantics are tested in detail in the test_tags module.
249 def test_type(self, db):
250 assert isinstance(db.tags, collections.abc.Set)
252 def test_none(self, db):
253 itags = iter(db.tags)
254 with pytest.raises(StopIteration):
256 assert len(db.tags) == 0
259 def test_some(self, db, maildir):
260 _, pathname = maildir.deliver()
261 msg, _ = db.add(pathname, sync_flags=False)
262 msg.tags.add('hello')
263 itags = iter(db.tags)
264 assert next(itags) == 'hello'
265 with pytest.raises(StopIteration):
267 assert 'hello' in msg.tags
269 def test_cache(self, db):
270 assert db.tags is db.tags
272 def test_iters(self, db):
281 def db(self, maildir, notmuch):
282 """Return a read-only notmuch2.Database.
284 The database will have 3 messages, 2 threads.
286 msgid, _ = maildir.deliver(body='foo')
287 maildir.deliver(body='bar')
288 maildir.deliver(body='baz',
289 headers=[('In-Reply-To', '<{}>'.format(msgid))])
291 with dbmod.Database(maildir.path, 'rw') as db:
294 def test_count_messages(self, db):
295 assert db.count_messages('*') == 3
297 def test_messages_type(self, db):
298 msgs = db.messages('*')
299 assert isinstance(msgs, collections.abc.Iterator)
301 def test_message_no_results(self, db):
302 msgs = db.messages('not_a_matching_query')
303 with pytest.raises(StopIteration):
306 def test_message_match(self, db):
307 msgs = db.messages('*')
309 assert isinstance(msg, notmuch2.Message)
311 def test_count_threads(self, db):
312 assert db.count_threads('*') == 2
314 def test_threads_type(self, db):
315 threads = db.threads('*')
316 assert isinstance(threads, collections.abc.Iterator)
318 def test_threads_no_match(self, db):
319 threads = db.threads('not_a_matching_query')
320 with pytest.raises(StopIteration):
323 def test_threads_match(self, db):
324 threads = db.threads('*')
325 thread = next(threads)
326 assert isinstance(thread, notmuch2.Thread)