tests: Remove classes from tests (sqlite)

tests/test_sqlite.py

@@ -1,12 +1,26 @@
 import datetime
-import unittest
+
+import pytest
 
 from scrapyd.jobstorage import Job
 from scrapyd.sqlite import JsonSqliteDict, JsonSqlitePriorityQueue, SqliteFinishedJobs
 
 
-class JsonSqliteDictTest(unittest.TestCase):
-    dict_class = JsonSqliteDict
+@pytest.fixture()
+def jsonsqlitepriorityqueue():
+    return JsonSqlitePriorityQueue()
+
+
+@pytest.fixture()
+def sqlitefinishedjobs():
+    q = SqliteFinishedJobs(":memory:")
+    q.add(Job("p1", "s1", end_time=datetime.datetime(2001, 2, 3, 4, 5, 6, 7)))
+    q.add(Job("p2", "s2", end_time=datetime.datetime(2001, 2, 3, 4, 5, 6, 8)))
+    q.add(Job("p3", "s3", end_time=datetime.datetime(2001, 2, 3, 4, 5, 6, 9)))
+    return q
+
+
+def test_jsonsqlitedict_basic_types():
     test_dict = (
         ("hello", "world"),
         ("int", 1),
@@ -16,170 +30,166 @@ class JsonSqliteDictTest(unittest.TestCase):
         ("dict", {"some": "dict"}),
     )
 
-    def test_basic_types(self):
-        test = self.test_dict
-        d = self.dict_class()
-        d.update(test)
+    d = JsonSqliteDict()
+    d.update(test_dict)
+
+    assert list(d.items()) == list(test_dict)
+
+    d.clear()
+
+    assert not d.items()
+
+
+def test_jsonsqlitedict_in():
+    d = JsonSqliteDict()
+
+    assert "test" not in d
+
+    d["test"] = 123
+
+    assert "test" in d
+
+
+def test_jsonsqlitedict_keyerror():
+    d = JsonSqliteDict()
+
+    with pytest.raises(KeyError):
+        d.__getitem__("test")
+
+
+def test_jsonsqlitedict_replace():
+    d = JsonSqliteDict()
+
+    assert d.get("test") is None
+
+    d["test"] = 123
 
-        self.assertEqual(list(d.items()), list(test))
+    assert d.get("test") == 123
 
-        d.clear()
+    d["test"] = 456
 
-        self.assertFalse(d.items())
+    assert d.get("test") == 456
 
-    def test_in(self):
-        d = self.dict_class()
 
-        self.assertNotIn("test", d)
+def test_jsonsqlitepriorityqueue_empty(jsonsqlitepriorityqueue):
+    assert jsonsqlitepriorityqueue.pop() is None
 
-        d["test"] = 123
 
-        self.assertIn("test", d)
+def test_jsonsqlitepriorityqueue_one(jsonsqlitepriorityqueue):
+    msg = "a message"
+    jsonsqlitepriorityqueue.put(msg)
 
-    def test_keyerror(self):
-        d = self.dict_class()
+    assert "_id" not in msg
+    assert jsonsqlitepriorityqueue.pop() == msg
+    assert jsonsqlitepriorityqueue.pop() is None
 
-        self.assertRaises(KeyError, d.__getitem__, "test")
 
-    def test_replace(self):
-        d = self.dict_class()
+def test_jsonsqlitepriorityqueue_multiple(jsonsqlitepriorityqueue):
+    msg1 = "first message"
+    msg2 = "second message"
+    jsonsqlitepriorityqueue.put(msg1)
+    jsonsqlitepriorityqueue.put(msg2)
+    out = []
+    out.append(jsonsqlitepriorityqueue.pop())
+    out.append(jsonsqlitepriorityqueue.pop())
 
-        self.assertEqual(d.get("test"), None)
+    assert msg1 in out
+    assert msg2 in out
+    assert jsonsqlitepriorityqueue.pop() is None
 
-        d["test"] = 123
 
-        self.assertEqual(d.get("test"), 123)
+def test_jsonsqlitepriorityqueue_priority(jsonsqlitepriorityqueue):
+    msg1 = "message 1"
+    msg2 = "message 2"
+    msg3 = "message 3"
+    msg4 = "message 4"
+    jsonsqlitepriorityqueue.put(msg1, priority=1.0)
+    jsonsqlitepriorityqueue.put(msg2, priority=5.0)
+    jsonsqlitepriorityqueue.put(msg3, priority=3.0)
+    jsonsqlitepriorityqueue.put(msg4, priority=2.0)
 
-        d["test"] = 456
+    assert jsonsqlitepriorityqueue.pop() == msg2
+    assert jsonsqlitepriorityqueue.pop() == msg3
+    assert jsonsqlitepriorityqueue.pop() == msg4
+    assert jsonsqlitepriorityqueue.pop() == msg1
 
-        self.assertEqual(d.get("test"), 456)
 
+def test_jsonsqlitepriorityqueue_iter_len_clear(jsonsqlitepriorityqueue):
+    assert len(jsonsqlitepriorityqueue) == 0
+    assert list(jsonsqlitepriorityqueue) == []
 
-class JsonSqlitePriorityQueueTest(unittest.TestCase):
-    queue_class = JsonSqlitePriorityQueue
+    msg1 = "message 1"
+    msg2 = "message 2"
+    msg3 = "message 3"
+    msg4 = "message 4"
+    jsonsqlitepriorityqueue.put(msg1, priority=1.0)
+    jsonsqlitepriorityqueue.put(msg2, priority=5.0)
+    jsonsqlitepriorityqueue.put(msg3, priority=3.0)
+    jsonsqlitepriorityqueue.put(msg4, priority=2.0)
 
-    supported_values = (
+    assert len(jsonsqlitepriorityqueue) == 4
+    assert list(jsonsqlitepriorityqueue) == [(msg2, 5.0), (msg3, 3.0), (msg4, 2.0), (msg1, 1.0)]
+
+    jsonsqlitepriorityqueue.clear()
+
+    assert len(jsonsqlitepriorityqueue) == 0
+    assert list(jsonsqlitepriorityqueue) == []
+
+
+def test_jsonsqlitepriorityqueue_remove(jsonsqlitepriorityqueue):
+    assert len(jsonsqlitepriorityqueue) == 0
+    assert list(jsonsqlitepriorityqueue) == []
+
+    msg1 = "good message 1"
+    msg2 = "bad message 2"
+    msg3 = "good message 3"
+    msg4 = "bad message 4"
+    jsonsqlitepriorityqueue.put(msg1)
+    jsonsqlitepriorityqueue.put(msg2)
+    jsonsqlitepriorityqueue.put(msg3)
+    jsonsqlitepriorityqueue.put(msg4)
+    jsonsqlitepriorityqueue.remove(lambda x: x.startswith("bad"))
+
+    assert list(jsonsqlitepriorityqueue) == [(msg1, 0.0), (msg3, 0.0)]
+
+
+@pytest.mark.parametrize(
+    "value",
+    [
         "native ascii str",
         "\xa3",
         123,
         1.2,
         True,
         ["a", "list", 1],
         {"a": "dict"},
-    )
+    ],
+)
+def test_jsonsqlitepriorityqueue_types(jsonsqlitepriorityqueue, value):
+    jsonsqlitepriorityqueue.put(value)
+
+    assert jsonsqlitepriorityqueue.pop() == value
+
+
+def test_sqlitefinishedjobs_add(sqlitefinishedjobs):
+    assert len(sqlitefinishedjobs) == 3
+
+
+def test_sqlitefinishedjobs_clear_all(sqlitefinishedjobs):
+    sqlitefinishedjobs.clear()
+
+    assert len(sqlitefinishedjobs) == 0
+
+
+def test_sqlitefinishedjobs_clear_keep_2(sqlitefinishedjobs):
+    sqlitefinishedjobs.clear(finished_to_keep=2)
+
+    assert len(sqlitefinishedjobs) == 2
+
+
+def test_sqlitefinishedjobs__iter__(sqlitefinishedjobs):
+    actual = list(sqlitefinishedjobs)
 
-    def setUp(self):
-        self.q = self.queue_class()
-
-    def test_empty(self):
-        self.assertIs(self.q.pop(), None)
-
-    def test_one(self):
-        msg = "a message"
-        self.q.put(msg)
-
-        self.assertNotIn("_id", msg)
-        self.assertEqual(self.q.pop(), msg)
-        self.assertIs(self.q.pop(), None)
-
-    def test_multiple(self):
-        msg1 = "first message"
-        msg2 = "second message"
-        self.q.put(msg1)
-        self.q.put(msg2)
-        out = []
-        out.append(self.q.pop())
-        out.append(self.q.pop())
-
-        self.assertIn(msg1, out)
-        self.assertIn(msg2, out)
-        self.assertIs(self.q.pop(), None)
-
-    def test_priority(self):
-        msg1 = "message 1"
-        msg2 = "message 2"
-        msg3 = "message 3"
-        msg4 = "message 4"
-        self.q.put(msg1, priority=1.0)
-        self.q.put(msg2, priority=5.0)
-        self.q.put(msg3, priority=3.0)
-        self.q.put(msg4, priority=2.0)
-
-        self.assertEqual(self.q.pop(), msg2)
-        self.assertEqual(self.q.pop(), msg3)
-        self.assertEqual(self.q.pop(), msg4)
-        self.assertEqual(self.q.pop(), msg1)
-
-    def test_iter_len_clear(self):
-        self.assertEqual(len(self.q), 0)
-        self.assertEqual(list(self.q), [])
-
-        msg1 = "message 1"
-        msg2 = "message 2"
-        msg3 = "message 3"
-        msg4 = "message 4"
-        self.q.put(msg1, priority=1.0)
-        self.q.put(msg2, priority=5.0)
-        self.q.put(msg3, priority=3.0)
-        self.q.put(msg4, priority=2.0)
-
-        self.assertEqual(len(self.q), 4)
-        self.assertEqual(list(self.q), [(msg2, 5.0), (msg3, 3.0), (msg4, 2.0), (msg1, 1.0)])
-
-        self.q.clear()
-
-        self.assertEqual(len(self.q), 0)
-        self.assertEqual(list(self.q), [])
-
-    def test_remove(self):
-        self.assertEqual(len(self.q), 0)
-        self.assertEqual(list(self.q), [])
-
-        msg1 = "good message 1"
-        msg2 = "bad message 2"
-        msg3 = "good message 3"
-        msg4 = "bad message 4"
-        self.q.put(msg1)
-        self.q.put(msg2)
-        self.q.put(msg3)
-        self.q.put(msg4)
-        self.q.remove(lambda x: x.startswith("bad"))
-
-        self.assertEqual(list(self.q), [(msg1, 0.0), (msg3, 0.0)])
-
-    def test_types(self):
-        for x in self.supported_values:
-            self.q.put(x)
-
-            self.assertEqual(self.q.pop(), x)
-
-
-class SqliteFinishedJobsTest(unittest.TestCase):
-    def setUp(self):
-        self.q = SqliteFinishedJobs(":memory:")
-        self.j1 = Job("p1", "s1", end_time=datetime.datetime(2001, 2, 3, 4, 5, 6, 7))
-        self.j2 = Job("p2", "s2", end_time=datetime.datetime(2001, 2, 3, 4, 5, 6, 8))
-        self.j3 = Job("p3", "s3", end_time=datetime.datetime(2001, 2, 3, 4, 5, 6, 9))
-        self.q.add(self.j1)
-        self.q.add(self.j2)
-        self.q.add(self.j3)
-
-    def test_add(self):
-        self.assertEqual(len(self.q), 3)
-
-    def test_clear_all(self):
-        self.q.clear()
-
-        self.assertEqual(len(self.q), 0)
-
-    def test_clear_keep_2(self):
-        self.q.clear(finished_to_keep=2)
-
-        self.assertEqual(len(self.q), 2)
-
-    def test__iter__(self):
-        actual = list(self.q)
-
-        self.assertEqual((actual[0][0], actual[0][1]), ("p3", "s3"))
-        self.assertEqual((actual[1][0], actual[1][1]), ("p2", "s2"))
-        self.assertEqual((actual[2][0], actual[2][1]), ("p1", "s1"))
+    assert (actual[0][0], actual[0][1]) == ("p3", "s3")
+    assert (actual[1][0], actual[1][1]) == ("p2", "s2")
+    assert (actual[2][0], actual[2][1]) == ("p1", "s1")