Changeset View
Changeset View
Standalone View
Standalone View
swh/indexer/tests/storage/test_storage.py
| Show All 14 Lines | |||||
| from swh.core.tests.db_testing import SingleDbTestFixture | from swh.core.tests.db_testing import SingleDbTestFixture | ||||
| from swh.indexer.tests.storage.generate_data_test import ( | from swh.indexer.tests.storage.generate_data_test import ( | ||||
| gen_content_mimetypes, gen_content_fossology_licenses | gen_content_mimetypes, gen_content_fossology_licenses | ||||
| ) | ) | ||||
| from swh.indexer.tests.storage import SQL_DIR | from swh.indexer.tests.storage import SQL_DIR | ||||
| @pytest.mark.db | @pytest.mark.db | ||||
| class BaseTestStorage(SingleDbTestFixture): | class BasePgTestStorage(SingleDbTestFixture): | ||||
| """Base test class for most indexer tests. | """Base test class for most indexer tests. | ||||
| It adds support for Storage testing to the SingleDbTestFixture class. | It adds support for Storage testing to the SingleDbTestFixture class. | ||||
| It will also build the database from the swh-indexed/sql/*.sql files. | It will also build the database from the swh-indexed/sql/*.sql files. | ||||
| """ | """ | ||||
| TEST_DB_NAME = 'softwareheritage-test-indexer' | TEST_DB_NAME = 'softwareheritage-test-indexer' | ||||
| TEST_DB_DUMP = os.path.join(SQL_DIR, '*.sql') | TEST_DB_DUMP = os.path.join(SQL_DIR, '*.sql') | ||||
| ▲ Show 20 Lines • Show All 43 Lines • ▼ Show 20 Lines | class BasePgTestStorage(SingleDbTestFixture): | ||||
| def reset_storage_tables(self): | def reset_storage_tables(self): | ||||
| excluded = {'indexer_configuration'} | excluded = {'indexer_configuration'} | ||||
| self.reset_db_tables(self.TEST_DB_NAME, excluded=excluded) | self.reset_db_tables(self.TEST_DB_NAME, excluded=excluded) | ||||
| db = self.test_db[self.TEST_DB_NAME] | db = self.test_db[self.TEST_DB_NAME] | ||||
| db.conn.commit() | db.conn.commit() | ||||
| @pytest.mark.db | class CommonTestStorage: | ||||
| class CommonTestStorage(BaseTestStorage): | |||||
| """Base class for Indexer Storage testing. | """Base class for Indexer Storage testing. | ||||
| """ | """ | ||||
| def test_check_config(self): | def test_check_config(self): | ||||
| self.assertTrue(self.storage.check_config(check_write=True)) | self.assertTrue(self.storage.check_config(check_write=True)) | ||||
| self.assertTrue(self.storage.check_config(check_write=False)) | self.assertTrue(self.storage.check_config(check_write=False)) | ||||
| ▲ Show 20 Lines • Show All 1,609 Lines • ▼ Show 20 Lines | |||||
| def test_indexer_configuration_get(self): | def test_indexer_configuration_get(self): | ||||
| tool = { | tool = { | ||||
| 'tool_name': 'nomos', | 'tool_name': 'nomos', | ||||
| 'tool_version': '3.1.0rc2-31-ga2cbb8c', | 'tool_version': '3.1.0rc2-31-ga2cbb8c', | ||||
| 'tool_configuration': {"command_line": "nomossa <filepath>"}, | 'tool_configuration': {"command_line": "nomossa <filepath>"}, | ||||
| } | } | ||||
| self.storage.indexer_configuration_add([tool]) | |||||
| actual_tool = self.storage.indexer_configuration_get(tool) | actual_tool = self.storage.indexer_configuration_get(tool) | ||||
| expected_tool = tool.copy() | expected_tool = tool.copy() | ||||
| expected_tool['id'] = 1 | del actual_tool['id'] | ||||
| self.assertEqual(expected_tool, actual_tool) | self.assertEqual(expected_tool, actual_tool) | ||||
| def test_indexer_configuration_metadata_get_missing_context(self): | def test_indexer_configuration_metadata_get_missing_context(self): | ||||
| tool = { | tool = { | ||||
| 'tool_name': 'swh-metadata-translator', | 'tool_name': 'swh-metadata-translator', | ||||
| 'tool_version': '0.0.1', | 'tool_version': '0.0.1', | ||||
| 'tool_configuration': {"context": "unknown-context"}, | 'tool_configuration': {"context": "unknown-context"}, | ||||
| } | } | ||||
| actual_tool = self.storage.indexer_configuration_get(tool) | actual_tool = self.storage.indexer_configuration_get(tool) | ||||
| self.assertIsNone(actual_tool) | self.assertIsNone(actual_tool) | ||||
| def test_indexer_configuration_metadata_get(self): | def test_indexer_configuration_metadata_get(self): | ||||
| tool = { | tool = { | ||||
| 'tool_name': 'swh-metadata-translator', | 'tool_name': 'swh-metadata-translator', | ||||
| 'tool_version': '0.0.1', | 'tool_version': '0.0.1', | ||||
| 'tool_configuration': {"type": "local", "context": "NpmMapping"}, | 'tool_configuration': {"type": "local", "context": "NpmMapping"}, | ||||
| } | } | ||||
| self.storage.indexer_configuration_add([tool]) | |||||
| actual_tool = self.storage.indexer_configuration_get(tool) | actual_tool = self.storage.indexer_configuration_get(tool) | ||||
| expected_tool = tool.copy() | expected_tool = tool.copy() | ||||
| expected_tool['id'] = actual_tool['id'] | expected_tool['id'] = actual_tool['id'] | ||||
| self.assertEqual(expected_tool, actual_tool) | self.assertEqual(expected_tool, actual_tool) | ||||
| @pytest.mark.property_based | @pytest.mark.property_based | ||||
| class PropBasedTestStorage(BaseTestStorage, unittest.TestCase): | class CommonPropBasedTestStorage(BasePgTestStorage): | ||||
| """Properties-based tests | """Properties-based tests | ||||
| """ | """ | ||||
| def test_generate_content_mimetype_get_range_limit_none(self): | def test_generate_content_mimetype_get_range_limit_none(self): | ||||
| """mimetype_get_range call with wrong limit input should fail""" | """mimetype_get_range call with wrong limit input should fail""" | ||||
| with self.assertRaises(ValueError) as e: | with self.assertRaises(ValueError) as e: | ||||
| self.storage.content_mimetype_get_range( | self.storage.content_mimetype_get_range( | ||||
| start=None, end=None, indexer_configuration_id=None, | start=None, end=None, indexer_configuration_id=None, | ||||
| ▲ Show 20 Lines • Show All 199 Lines • ▼ Show 20 Lines | def test_generate_fossology_license_get_range_limit( | ||||
| actual_ids2 = actual_results2['ids'] | actual_ids2 = actual_results2['ids'] | ||||
| actual_next2 = actual_results2['next'] | actual_next2 = actual_results2['next'] | ||||
| self.assertIsNone(actual_next2) | self.assertIsNone(actual_next2) | ||||
| expected_fossology_licenses2 = [content_ids[-1]] | expected_fossology_licenses2 = [content_ids[-1]] | ||||
| self.assertEqual(expected_fossology_licenses2, actual_ids2) | self.assertEqual(expected_fossology_licenses2, actual_ids2) | ||||
| class IndexerTestStorage(CommonTestStorage, unittest.TestCase): | @pytest.mark.db | ||||
| class IndexerTestStorage(CommonTestStorage, BasePgTestStorage, | |||||
| unittest.TestCase): | |||||
| """Running the tests locally. | """Running the tests locally. | ||||
| For the client api tests (remote storage), see | For the client api tests (remote storage), see | ||||
| `class`:swh.indexer.storage.test_api_client:TestRemoteStorage | `class`:swh.indexer.storage.test_api_client:TestRemoteStorage | ||||
| class. | class. | ||||
| """ | """ | ||||
| pass | pass | ||||
| @pytest.mark.property_based | |||||
| class PropIndexerTestStorage(CommonPropBasedTestStorage, BasePgTestStorage, | |||||
| unittest.TestCase): | |||||
| pass | |||||