How to use the perceval.archive.Archive function in perceval

To help you get started, we’ve selected a few perceval examples, based on popular ways it is used in public projects.

Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.

github chaoss / grimoirelab-perceval / tests / test_archive.py View on Github external
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)
github chaoss / grimoirelab-perceval / tests / test_archive.py View on Github external
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)
github chaoss / grimoirelab-perceval / tests / test_archive.py View on Github external
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)
github chaoss / grimoirelab-perceval / tests / test_archive.py View on Github external
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)
github chaoss / grimoirelab-perceval / tests / test_client.py View on Github external
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)
github chaoss / grimoirelab-perceval / tests / test_archive.py View on Github external
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})
github chaoss / grimoirelab-perceval / perceval / archive.py View on Github external
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)
github chaoss / grimoirelab-perceval / perceval / backend.py View on Github external
: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))
github chaoss / grimoirelab-perceval / perceval / archive.py View on Github external
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