Draw Process descendant trees, and guard against recursion within them.

[plomtask] / tests / processes.py

diff --git a/tests/processes.py b/tests/processes.py
index e8967f70273e2345df966d599d50fdbeb4069e39..02f664477fa5147b0396538e9a63263dca9ce73a 100644 (file)
--- a/tests/processes.py
+++ b/tests/processes.py
@@ -1,6 +1,5 @@
 """Test Processes module."""
 from unittest import TestCase
 """Test Processes module."""
 from unittest import TestCase

-from urllib.parse import urlencode

 from tests.utils import TestCaseWithDB, TestCaseWithServer
 from plomtask.processes import Process
 from plomtask.exceptions import NotFoundException, BadFormatException
 from tests.utils import TestCaseWithDB, TestCaseWithServer
 from plomtask.processes import Process
 from plomtask.exceptions import NotFoundException, BadFormatException


@@ -37,6 +36,19 @@ class TestsWithDB(TestCaseWithDB):
         p_saved.save(self.db_conn)
         p_loaded = Process.by_id(self.db_conn, p_saved.id_)
         self.assertEqual(p_saved.title.history, p_loaded.title.history)
         p_saved.save(self.db_conn)
         p_loaded = Process.by_id(self.db_conn, p_saved.id_)
         self.assertEqual(p_saved.title.history, p_loaded.title.history)

+        p_9 = Process(9)
+        p_9.child_ids = [4]
+        with self.assertRaises(NotFoundException):
+            p_9.save(self.db_conn)
+        p_9.child_ids = [5]
+        p_9.save(self.db_conn)
+        p_5 = Process.by_id(self.db_conn, 5)
+        p_5.child_ids = [1]
+        p_5.save(self.db_conn)
+        p_1 = Process.by_id(self.db_conn, 1)
+        p_1.child_ids = [9]
+        with self.assertRaises(BadFormatException):
+            p_1.save(self.db_conn)

 
     def test_Process_by_id(self) -> None:
         """Test Process.by_id()."""
 
     def test_Process_by_id(self) -> None:
         """Test Process.by_id()."""


@@ -68,36 +80,47 @@ class TestsWithServer(TestCaseWithServer):
 
     def test_do_POST_process(self) -> None:
         """Test POST /process and its effect on the database."""
 
     def test_do_POST_process(self) -> None:
         """Test POST /process and its effect on the database."""

-        def post_data_to_expect(form_data: dict[str, object],
-                                to_: str, expect: int) -> None:
-            encoded_form_data = urlencode(form_data).encode('utf-8')
-            headers = {'Content-Type': 'application/x-www-form-urlencoded',
-                       'Content-Length': str(len(encoded_form_data))}
-            self.conn.request('POST', to_,
-                              body=encoded_form_data, headers=headers)
-            self.assertEqual(self.conn.getresponse().status, expect)
-        form_data = {'title': 'foo', 'description': 'foo', 'effort': 1.0}
-        post_data_to_expect(form_data, '/process?id=FOO', 400)
+        form_data = {'title': 'foo', 'description': 'foo', 'effort': 1.1}
+        self.check_post(form_data, '/process?id=', 302, '/')
+        self.check_post(form_data, '/process?id=FOO', 400)

         form_data['effort'] = 'foo'
         form_data['effort'] = 'foo'

-        post_data_to_expect(form_data, '/process?id=', 400)
-        form_data['effort'] = None
-        post_data_to_expect(form_data, '/process?id=', 400)
-        form_data = {'title': None, 'description': 1, 'effort': 1.0}
-        post_data_to_expect(form_data, '/process?id=', 302)
-        retrieved = Process.by_id(self.db_conn, 1)
-        self.assertEqual(retrieved.title.newest, 'None')
+        self.check_post(form_data, '/process?id=', 400)
+        self.check_post({}, '/process?id=', 400)
+        form_data = {'title': '', 'description': ''}
+        self.check_post(form_data, '/process?id=', 400)
+        form_data = {'title': '', 'effort': 1.1}
+        self.check_post(form_data, '/process?id=', 400)
+        form_data = {'description': '', 'effort': 1.0}
+        self.check_post(form_data, '/process?id=', 400)
+        form_data = {'title': '', 'description': '',
+                     'effort': 1.1, 'children': [1]}
+        self.check_post(form_data, '/process?id=', 302, '/')
+        form_data['children'] = 1.1
+        self.check_post(form_data, '/process?id=', 400)
+        form_data['children'] = 'a'
+        self.check_post(form_data, '/process?id=', 400)
+        form_data['children'] = [1.2]
+        self.check_post(form_data, '/process?id=', 400)
+        form_data['children'] = ['b']
+        self.check_post(form_data, '/process?id=', 400)
+        form_data['children'] = [2]
+        self.check_post(form_data, '/process?id=1', 400)
+        retrieved_1 = Process.by_id(self.db_conn, 1)
+        self.assertEqual(retrieved_1.title.newest, 'foo')
+        self.assertEqual(retrieved_1.child_ids, [])
+        retrieved_2 = Process.by_id(self.db_conn, 2)
+        self.assertEqual(retrieved_2.child_ids, [1])
+        form_data = {'title': 'bar', 'description': 'bar', 'effort': 1.1}
+        self.check_post(form_data, '/process?id=1', 302, '/')
+        retrieved_1 = Process.by_id(self.db_conn, 1)
+        self.assertEqual(retrieved_1.title.newest, 'bar')

         self.assertEqual([p.id_ for p in Process.all(self.db_conn)],
         self.assertEqual([p.id_ for p in Process.all(self.db_conn)],

-                         [retrieved.id_])
+                         [retrieved_1.id_, retrieved_2.id_])

 
     def test_do_GET(self) -> None:
         """Test /process and /processes response codes."""
 
     def test_do_GET(self) -> None:
         """Test /process and /processes response codes."""

-        self.conn.request('GET', '/process')
-        self.assertEqual(self.conn.getresponse().status, 200)
-        self.conn.request('GET', '/process?id=')
-        self.assertEqual(self.conn.getresponse().status, 200)
-        self.conn.request('GET', '/process?id=0')
-        self.assertEqual(self.conn.getresponse().status, 400)
-        self.conn.request('GET', '/process?id=FOO')
-        self.assertEqual(self.conn.getresponse().status, 400)
-        self.conn.request('GET', '/processes')
-        self.assertEqual(self.conn.getresponse().status, 200)
+        self.check_get('/process', 200)
+        self.check_get('/process?id=', 200)
+        self.check_get('/process?id=0', 400)
+        self.check_get('/process?id=FOO', 400)
+        self.check_get('/processes', 200)