diff --git a/swh/lister/debian/tests/conftest.py b/swh/lister/debian/tests/conftest.py index 4443ef8..865e9e1 100644 --- a/swh/lister/debian/tests/conftest.py +++ b/swh/lister/debian/tests/conftest.py @@ -1,80 +1,61 @@ # Copyright (C) 2019-2020 The Software Heritage developers # See the AUTHORS file at the top-level directory of this distribution # License: GNU General Public License version 3, or any later version # See top-level LICENSE file for more information import os import pytest -from pytest_postgresql.janitor import DatabaseJanitor -from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker from swh.core.db.pytest_plugin import postgresql_fact -from swh.lister.core.models import SQLBase from swh.lister.debian import debian_init import swh.scheduler SQL_DIR = os.path.join(os.path.dirname(swh.scheduler.__file__), "sql") postgresql_scheduler = postgresql_fact( "postgresql_proc", db_name="scheduler-lister", dump_files=os.path.join(SQL_DIR, "*.sql"), # do not truncate the task tables, it's required in between test no_truncate_tables={"dbversion", "priority_ratio", "task"}, ) @pytest.fixture def swh_scheduler_config(postgresql_scheduler): return {"db": postgresql_scheduler.dsn} @pytest.fixture def lister_under_test(): return "debian" @pytest.fixture def lister_debian(swh_lister): # Initialize the debian data model debian_init( swh_lister.db_engine, suites=["stretch"], components=["main", "contrib"] ) # Add the load-deb-package in the scheduler backend swh_lister.scheduler.create_task_type( { "type": "load-deb-package", "description": "Load a Debian package", - "backend_name": "swh.loader.debian.tasks.LoaderDebianPackage", + "backend_name": "swh.loader.packages.debian.tasks.LoaderDebianPackage", "default_interval": "1 day", } ) return swh_lister @pytest.fixture -def sqlalchemy_engine(postgresql_proc): - pg_host = postgresql_proc.host - pg_port = postgresql_proc.port - pg_user = postgresql_proc.user - - pg_db = "sqlalchemy-tests" - - url = f"postgresql://{pg_user}@{pg_host}:{pg_port}/{pg_db}" - with DatabaseJanitor(pg_user, pg_host, pg_port, pg_db, postgresql_proc.version): - engine = create_engine(url) - yield engine - engine.dispose() - - -@pytest.fixture -def session(sqlalchemy_engine): - SQLBase.metadata.create_all(sqlalchemy_engine) - Session = sessionmaker(bind=sqlalchemy_engine) - session = Session() +def session(lister_db_url, engine): + session = sessionmaker(bind=engine)() yield session session.close() + engine.dispose() diff --git a/swh/lister/launchpad/tests/conftest.py b/swh/lister/launchpad/tests/conftest.py index bd69f08..0fb42c5 100644 --- a/swh/lister/launchpad/tests/conftest.py +++ b/swh/lister/launchpad/tests/conftest.py @@ -1,52 +1,47 @@ # Copyright (C) 2020 The Software Heritage developers # See the AUTHORS file at the top-level directory of this distribution # License: GNU General Public License version 3, or any later version # See top-level LICENSE file for more information from datetime import datetime import json import os from unittest.mock import patch import pytest -from sqlalchemy.engine import create_engine from swh.lister import get_lister -from swh.lister.core.models import initialize @pytest.fixture -def lister_launchpad(datadir, lister_db_url, swh_scheduler): +def lister_launchpad(datadir, lister_db_url, engine, swh_scheduler): class Collection: entries = [] def __init__(self, file): self.entries = [Repo(r) for r in file] def __getitem__(self, key): return self.entries[key] class Repo: def __init__(self, d: dict): for key in d.keys(): if key == "date_last_modified": setattr(self, key, datetime.fromisoformat(d[key])) else: setattr(self, key, d[key]) def mock_lp_response(page) -> Collection: response_filepath = os.path.join(datadir, f"response{page}.json") with open(response_filepath, "r", encoding="utf-8") as f: return Collection(json.load(f)) with patch("launchpadlib.launchpad.Launchpad.login_anonymously"): lister = get_lister("launchpad", db_url=lister_db_url) - lister.scheduler = swh_scheduler # inject scheduler fixture lister.launchpad.git_repositories.getRepositories.side_effect = [ mock_lp_response(i) for i in range(3) ] - initialize(create_engine(lister_db_url), drop_tables=True) - return lister diff --git a/swh/lister/pytest_plugin.py b/swh/lister/pytest_plugin.py index 5a255b5..82dd08b 100644 --- a/swh/lister/pytest_plugin.py +++ b/swh/lister/pytest_plugin.py @@ -1,58 +1,62 @@ # Copyright (C) 2019-2020 The Software Heritage developers # See the AUTHORS file at the top-level directory of this distribution # License: GNU General Public License version 3, or any later version # See top-level LICENSE file for more information import logging import os import pytest from sqlalchemy import create_engine import yaml from swh.lister import SUPPORTED_LISTERS, get_lister from swh.lister.core.models import initialize logger = logging.getLogger(__name__) @pytest.fixture def lister_db_url(postgresql): db_params = postgresql.get_dsn_parameters() db_url = "postgresql://{user}@{host}:{port}/{dbname}".format(**db_params) logger.debug("lister db_url: %s", db_url) return db_url @pytest.fixture def lister_under_test(): """Fixture to determine which lister to test""" return "core" @pytest.fixture def swh_lister_config(lister_db_url, swh_scheduler_config): return { "scheduler": {"cls": "local", **swh_scheduler_config}, "lister": {"cls": "local", "args": {"db": lister_db_url},}, "credentials": {}, "cache_responses": False, } @pytest.fixture(autouse=True) def swh_config(swh_lister_config, monkeypatch, tmp_path): conf_path = os.path.join(str(tmp_path), "lister.yml") with open(conf_path, "w") as f: f.write(yaml.dump(swh_lister_config)) monkeypatch.setenv("SWH_CONFIG_FILENAME", conf_path) return conf_path @pytest.fixture -def swh_lister(lister_db_url, swh_scheduler, lister_under_test, swh_config): - assert lister_under_test in SUPPORTED_LISTERS - lister = get_lister(lister_under_test, db_url=lister_db_url) - initialize(create_engine(lister_db_url), drop_tables=True) +def engine(lister_db_url): + engine = create_engine(lister_db_url) + initialize(engine, drop_tables=True) + return engine + - return lister +@pytest.fixture +def swh_lister(engine, lister_db_url, lister_under_test, swh_config): + assert lister_under_test in SUPPORTED_LISTERS + return get_lister(lister_under_test, db_url=lister_db_url)