Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_init(self):
"""Test whether an archive is propertly initialized"""
archive_path = os.path.join(self.test_path, 'myarchive')
_ = Archive.create(archive_path)
archive = Archive(archive_path)
self.assertEqual(archive.archive_path, archive_path)
self.assertEqual(archive.created_on, None)
self.assertEqual(archive.origin, None)
self.assertEqual(archive.backend_name, None)
self.assertEqual(archive.backend_version, None)
self.assertEqual(archive.category, None)
self.assertEqual(archive.backend_params, None)
def test_init_not_valid_archive(self):
"""Test if an exception is raised when the file is an invalid archive"""
archive_path = os.path.join(self.test_path, 'invalid_archive')
with open(archive_path, 'w') as fd:
fd.write("Invalid archive file")
with self.assertRaisesRegex(ArchiveError, "invalid archive file"):
_ = Archive(archive_path)
def test_create(self):
"""Test a new an empty archive is created"""
archive_path = os.path.join(self.test_path, 'myarchive')
archive = Archive.create(archive_path)
# Archive file was created
self.assertEqual(archive.archive_path, archive_path)
self.assertEqual(os.path.exists(archive.archive_path), True)
# Properties are initialized
self.assertEqual(archive.created_on, None)
self.assertEqual(archive.origin, None)
self.assertEqual(archive.backend_name, None)
self.assertEqual(archive.backend_version, None)
self.assertEqual(archive.category, None)
self.assertEqual(archive.backend_params, None)
# Tables are empty
nrows = count_number_rows(archive_path, Archive.ARCHIVE_TABLE)
self.assertEqual(nrows, 0)
def test_create_archive(self, mock_uuid):
"""Test if a new archive is created"""
mock_uuid.return_value = MockUUID('AB0123456789')
archive_mng_path = os.path.join(self.test_path, ARCHIVE_TEST_DIR)
manager = ArchiveManager(archive_mng_path)
archive = manager.create_archive()
self.assertIsInstance(archive, Archive)
expected = os.path.join(archive_mng_path, 'AB', '0123456789.sqlite3')
self.assertEqual(archive.archive_path, expected)
self.assertEqual(os.path.exists(archive.archive_path), True)
def test_fetch_from_archive(self):
"""Test whether responses are correctly fecthed from an archive"""
archive_path = os.path.join(self.test_path, 'myarchive')
archive = Archive.create(archive_path)
httpretty.register_uri(httpretty.GET,
CLIENT_SUPERMAN_URL,
body="good",
status=200)
client = MockedClient(CLIENT_API_URL, sleep_time=0.1, max_retries=1, archive=archive)
answer_api = client.fetch(CLIENT_SUPERMAN_URL)
client = MockedClient(CLIENT_API_URL, sleep_time=0.1, max_retries=1, archive=archive, from_archive=True)
answer_archive = client.fetch(CLIENT_SUPERMAN_URL)
self.assertEqual(answer_api.text, answer_archive.text)
def test_init_metadata(self):
"""Test whether metadata information is properly initialized"""
archive_path = os.path.join(self.test_path, 'myarchive')
archive = Archive.create(archive_path)
before_dt = datetime_to_utc(datetime_utcnow())
archive.init_metadata('marvel.com', 'marvel-comics-backend', '0.1.0',
'issue', {'from_date': before_dt})
after_dt = datetime_to_utc(datetime_utcnow())
archive_copy = Archive(archive_path)
# Both copies should have the same parameters
for arch in [archive, archive_copy]:
self.assertEqual(arch.origin, 'marvel.com')
self.assertEqual(arch.backend_name, 'marvel-comics-backend')
self.assertEqual(arch.backend_version, '0.1.0')
self.assertEqual(arch.category, 'issue')
self.assertGreaterEqual(arch.created_on, before_dt)
self.assertLessEqual(arch.created_on, after_dt)
self.assertDictEqual(arch.backend_params, {'from_date': before_dt})
def remove_archive(self, archive_path):
"""Remove an archive.
This method deletes from the filesystem the archive stored
in `archive_path`.
:param archive_path: path to the archive
:raises ArchiveManangerError: when an error occurs removing the
archive
"""
try:
Archive(archive_path)
except ArchiveError as e:
raise ArchiveManagerError(cause=str(e))
os.remove(archive_path)
:param category: category of the items to retrieve
:param archived_after: return items archived after this date
:returns: a generator of archived items
"""
init_args = find_signature_parameters(backend_class.__init__,
backend_args)
backend = backend_class(**init_args)
filepaths = manager.search(backend.origin,
backend.__class__.__name__,
category,
archived_after)
for filepath in filepaths:
backend.archive = Archive(filepath)
items = backend.fetch_from_archive()
try:
for item in items:
yield item
except ArchiveError as e:
logger.warning("Ignoring %s archive due to: %s", filepath, str(e))
def _search_archives(self, origin, backend_name, category, archived_after):
"""Search archives using filters."""
for archive_path in self._search_files():
try:
archive = Archive(archive_path)
except ArchiveError:
continue
match = archive.origin == origin and \
archive.backend_name == backend_name and \
archive.category == category and \
archive.created_on >= archived_after
if not match:
continue
yield archive_path, archive.created_on