1 """Test Processes module."""
2 from unittest import TestCase
3 from urllib.parse import urlencode
4 from tests.utils import TestCaseWithDB, TestCaseWithServer
5 from plomtask.processes import Process
6 from plomtask.exceptions import NotFoundException
9 class TestsSansDB(TestCase):
10 """Module tests not requiring DB setup."""
12 def test_Process_versioned_defaults(self) -> None:
13 """Test defaults of Process' VersionedAttributes."""
14 self.assertEqual(Process(None).title.newest, 'UNNAMED')
15 self.assertEqual(Process(None).description.newest, '')
16 self.assertEqual(Process(None).effort.newest, 1.0)
19 class TestsWithDB(TestCaseWithDB):
20 """Mdule tests not requiring DB setup."""
22 def test_Process_save(self) -> None:
23 """Test Process.save()."""
24 p_saved = Process(None)
25 p_saved.save(self.db_conn)
26 self.assertEqual(p_saved.id_,
27 Process.by_id(self.db_conn, 1, create=False).id_)
29 p_saved.save(self.db_conn)
30 self.assertEqual(p_saved.id_,
31 Process.by_id(self.db_conn, 2, create=False).id_)
33 p_saved.save(self.db_conn)
34 self.assertEqual(p_saved.id_,
35 Process.by_id(self.db_conn, 5, create=False).id_)
36 p_saved.title.set('named')
37 p_loaded = Process.by_id(self.db_conn, p_saved.id_)
38 self.assertNotEqual(p_saved.title.history, p_loaded.title.history)
39 p_saved.save(self.db_conn)
40 p_loaded = Process.by_id(self.db_conn, p_saved.id_)
41 self.assertEqual(p_saved.title.history, p_loaded.title.history)
43 def test_Process_by_id(self) -> None:
44 """Test Process.by_id()."""
45 with self.assertRaises(NotFoundException):
46 Process.by_id(self.db_conn, None, create=False)
47 with self.assertRaises(NotFoundException):
48 Process.by_id(self.db_conn, 0, create=False)
49 with self.assertRaises(NotFoundException):
50 Process.by_id(self.db_conn, 1, create=False)
51 self.assertNotEqual(Process(1).id_,
52 Process.by_id(self.db_conn, None, create=True).id_)
53 self.assertNotEqual(Process(1).id_,
54 Process.by_id(self.db_conn, 0, create=True).id_)
55 self.assertEqual(Process(1).id_,
56 Process.by_id(self.db_conn, 1, create=True).id_)
57 self.assertEqual(Process(2).id_,
58 Process.by_id(self.db_conn, 2, create=True).id_)
60 def test_Process_all(self) -> None:
61 """Test Process.all()."""
63 p_1.save(self.db_conn)
65 p_2.save(self.db_conn)
66 self.assertEqual({p_1.id_, p_2.id_},
67 set(p.id_ for p in Process.all(self.db_conn)))
70 class TestsWithServer(TestCaseWithServer):
71 """Module tests against our HTTP server/handler (and database)."""
73 def test_do_POST_process(self) -> None:
74 """Test POST /process and its effect on the database."""
75 def post_data_to_expect(form_data: dict[str, object],
76 to_: str, expect: int) -> None:
77 encoded_form_data = urlencode(form_data).encode('utf-8')
78 headers = {'Content-Type': 'application/x-www-form-urlencoded',
79 'Content-Length': str(len(encoded_form_data))}
80 self.conn.request('POST', to_,
81 body=encoded_form_data, headers=headers)
82 self.assertEqual(self.conn.getresponse().status, expect)
83 form_data = {'title': 'foo', 'description': 'foo', 'effort': 1.0}
84 post_data_to_expect(form_data, '/process?id=FOO', 400)
85 form_data['effort'] = 'foo'
86 post_data_to_expect(form_data, '/process?id=', 400)
87 form_data['effort'] = None
88 post_data_to_expect(form_data, '/process?id=', 400)
89 form_data = {'title': None, 'description': 1, 'effort': 1.0}
90 post_data_to_expect(form_data, '/process?id=', 302)
91 retrieved = Process.by_id(self.db_conn, 1)
92 self.assertEqual(retrieved.title.newest, 'None')
93 self.assertEqual([p.id_ for p in Process.all(self.db_conn)],
96 def test_do_GET(self) -> None:
97 """Test /process and /processes response codes."""
98 self.conn.request('GET', '/process')
99 self.assertEqual(self.conn.getresponse().status, 200)
100 self.conn.request('GET', '/process?id=')
101 self.assertEqual(self.conn.getresponse().status, 200)
102 self.conn.request('GET', '/process?id=0')
103 self.assertEqual(self.conn.getresponse().status, 200)
104 self.conn.request('GET', '/process?id=FOO')
105 self.assertEqual(self.conn.getresponse().status, 400)
106 self.conn.request('GET', '/processes')
107 self.assertEqual(self.conn.getresponse().status, 200)