Changeset View
Changeset View
Standalone View
Standalone View
swh/scheduler/tests/updater/test_backend.py
| Show All 21 Lines | |||||
| @pytest.mark.db | @pytest.mark.db | ||||
| class SchedulerUpdaterBackendTest(SingleDbTestFixture, unittest.TestCase): | class SchedulerUpdaterBackendTest(SingleDbTestFixture, unittest.TestCase): | ||||
| TEST_DB_NAME = 'softwareheritage-scheduler-updater-test' | TEST_DB_NAME = 'softwareheritage-scheduler-updater-test' | ||||
| TEST_DB_DUMP = os.path.join(SQL_DIR, 'updater', '*.sql') | TEST_DB_DUMP = os.path.join(SQL_DIR, 'updater', '*.sql') | ||||
| def setUp(self): | def setUp(self): | ||||
| super().setUp() | super().setUp() | ||||
| config = { | config = { | ||||
| 'scheduling_updater_db': 'dbname=' + self.TEST_DB_NAME, | 'db': 'dbname=' + self.TEST_DB_NAME, | ||||
| 'cache_read_limit': 1000, | 'cache_read_limit': 1000, | ||||
| } | } | ||||
| self.backend = SchedulerUpdaterBackend(**config) | self.backend = SchedulerUpdaterBackend(**config) | ||||
| def _empty_tables(self): | def _empty_tables(self): | ||||
| self.cursor.execute( | self.cursor.execute( | ||||
| """SELECT table_name FROM information_schema.tables | """SELECT table_name FROM information_schema.tables | ||||
| WHERE table_schema = %s""", ('public', )) | WHERE table_schema = %s""", ('public', )) | ||||
| tables = set(table for (table,) in self.cursor.fetchall()) | tables = set(table for (table,) in self.cursor.fetchall()) | ||||
| for table in tables: | for table in tables: | ||||
| self.cursor.execute('truncate table %s cascade' % table) | self.cursor.execute('truncate table %s cascade' % table) | ||||
| self.conn.commit() | self.conn.commit() | ||||
| def tearDown(self): | def tearDown(self): | ||||
| self.backend.close_connection() | |||||
| self._empty_tables() | self._empty_tables() | ||||
| super().tearDown() | super().tearDown() | ||||
| @given(sets( | @given(sets( | ||||
| from_regex( | from_regex( | ||||
| r'^https://somewhere[.]org/[a-z0-9]{5,7}/[a-z0-9]{3,10}$'), | r'^https://somewhere[.]org/[a-z0-9]{5,7}/[a-z0-9]{3,10}$'), | ||||
| min_size=10, max_size=15)) | min_size=10, max_size=15)) | ||||
| def test_cache_read(self, urls): | def test_cache_read(self, urls): | ||||
| def gen_events(urls): | def gen_events(urls): | ||||
| for url in urls: | for url in urls: | ||||
| yield SWHEvent({ | yield SWHEvent({ | ||||
| 'url': url, | 'url': url, | ||||
| 'type': 'create', | 'type': 'create', | ||||
| 'origin_type': 'git', | 'origin_type': 'git', | ||||
| }) | }) | ||||
| self.backend.cache_put(gen_events(urls)) | self.backend.cache_put(gen_events(urls)) | ||||
| r = self.backend.cache_read(timestamp=utcnow()) | r = self.backend.cache_read(timestamp=utcnow()) | ||||
| self.assertNotEqual(r, []) | self.assertNotEqual(r, []) | ||||