Skip to content

Commit

Permalink
tests: Remove classes from tests (jobstorage)
Browse files Browse the repository at this point in the history
  • Loading branch information
jpmckinney committed Jul 19, 2024
1 parent e3a35f6 commit 1699ed7
Showing 1 changed file with 44 additions and 40 deletions.
84 changes: 44 additions & 40 deletions tests/test_jobstorage.py
Original file line number Diff line number Diff line change
@@ -1,60 +1,64 @@
from twisted.trial import unittest
import pytest
from zope.interface.verify import verifyObject

from scrapyd.config import Config
from scrapyd.interfaces import IJobStorage
from scrapyd.jobstorage import Job, MemoryJobStorage, SqliteJobStorage

j1, j2, j3 = Job("p1", "s1"), Job("p2", "s2"), Job("p3", "s3")
j1 = Job("p1", "s1")
j2 = Job("p2", "s2")
j3 = Job("p3", "s3")


class MemoryJobStorageTest(unittest.TestCase):
def setUp(self):
d = self.mktemp()
config = Config(values={"dbs_dir": d, "finished_to_keep": "2"})
self.jobst = MemoryJobStorage(config)
self.j1, self.j2, self.j3 = j1, j2, j3
self.jobst.add(self.j1)
self.jobst.add(self.j2)
self.jobst.add(self.j3)
@pytest.fixture()
def sqlitejobstorage(tmpdir):
return SqliteJobStorage(Config(values={"dbs_dir": tmpdir, "finished_to_keep": "2"}))

def test_interface(self):
verifyObject(IJobStorage, self.jobst)

def test_add(self):
self.assertEqual(len(self.jobst.list()), 2)
@pytest.fixture()
def memoryjobstorage(tmpdir):
storage = MemoryJobStorage(Config(values={"dbs_dir": tmpdir, "finished_to_keep": "2"}))
storage.add(j1)
storage.add(j2)
storage.add(j3)
return storage

def test_iter(self):
actual = list(self.jobst)

self.assertEqual(actual[0], j2)
self.assertEqual(actual[1], j3)
self.assertEqual(len(actual), 2)
def test_memory_interface(memoryjobstorage):
verifyObject(IJobStorage, memoryjobstorage)

def test_len(self):
self.assertEqual(len(self.jobst), 2)

def test_memory_add(memoryjobstorage):
assert len(memoryjobstorage.list()) == 2

class SqliteJobsStorageTest(unittest.TestCase):
def setUp(self):
d = self.mktemp()
config = Config(values={"dbs_dir": d, "finished_to_keep": "2"})
self.jobst = SqliteJobStorage(config)
self.j1, self.j2, self.j3 = j1, j2, j3

def test_interface(self):
verifyObject(IJobStorage, self.jobst)
def test_memory_iter(memoryjobstorage):
actual = list(memoryjobstorage)

def test_add(self):
self.jobst.add(self.j1)
self.jobst.add(self.j2)
self.jobst.add(self.j3)
assert actual[0] == j2
assert actual[1] == j3
assert len(actual) == 2

self.assertEqual(len(self.jobst.list()), 2)

def test_iter(self):
self.jobst.add(self.j1)
self.jobst.add(self.j2)
self.jobst.add(self.j3)
def test_len(memoryjobstorage):
assert len(memoryjobstorage) == 2

self.assertEqual(len(self.jobst), 2)

def test_sqlite_interface(sqlitejobstorage):
verifyObject(IJobStorage, sqlitejobstorage)


def test_sqlite_add(sqlitejobstorage):
sqlitejobstorage.add(j1)
sqlitejobstorage.add(j2)
sqlitejobstorage.add(j3)

assert len(sqlitejobstorage.list()) == 2


def test_sqlite_iter(sqlitejobstorage):
sqlitejobstorage.add(j1)
sqlitejobstorage.add(j2)
sqlitejobstorage.add(j3)

assert len(sqlitejobstorage) == 2

0 comments on commit 1699ed7

Please sign in to comment.