Page MenuHomeSoftware Heritage
Differential D2799 Diff 10543 swh/lister/core/tests/conftest.py

Changeset View

Standalone View

View Options

swh/lister/core/tests/conftest.py

# Copyright (C) 2019 The Software Heritage developers # Copyright (C) 2019 The Software Heritage developers
# See the AUTHORS file at the top-level directory of this distribution # See the AUTHORS file at the top-level directory of this distribution
# License: GNU General Public License version 3, or any later version # License: GNU General Public License version 3, or any later version
# See top-level LICENSE file for more information # See top-level LICENSE file for more information
from swh.scheduler.tests.conftest import * # noqa from swh.scheduler.tests.conftest import * # noqa
import logging import logging
import pytest import pytest
from sqlalchemy import create_engine from sqlalchemy import create_engine
from swh.lister import get_lister, SUPPORTED_LISTERS from swh.lister import get_lister, SUPPORTED_LISTERS
from swh.lister.core.models import initialize from swh.lister.core.models import initialize
logger = logging.getLogger(__name__) logger = logging.getLogger(__name__)
@pytest.fixture @pytest.fixture
def swh_listers(request, postgresql_proc, postgresql, swh_scheduler): def lister_db_url(postgresql_proc, postgresql):
db_url = 'postgresql://{user}@{host}:{port}/{dbname}'.format( db_url = 'postgresql://{user}@{host}:{port}/{dbname}'.format(
host=postgresql_proc.host, host=postgresql_proc.host,
port=postgresql_proc.port, port=postgresql_proc.port,
user='postgres', user='postgres',
dbname='tests') dbname='tests')
logger.debug('lister db_url: %s', db_url) logger.debug('lister db_url: %s', db_url)
return db_url
@pytest.fixture
def swh_listers(request, lister_db_url, swh_scheduler):
listers = {} listers = {}
# Prepare schema for all listers # Prepare schema for all listers
for lister_name in SUPPORTED_LISTERS: for lister_name in SUPPORTED_LISTERS:
lister = get_lister(lister_name, db_url=db_url) lister = get_lister(lister_name, db_url=lister_db_url)
lister.scheduler = swh_scheduler # inject scheduler fixture lister.scheduler = swh_scheduler # inject scheduler fixture
listers[lister_name] = lister listers[lister_name] = lister
ardumontUnsubmitted
Not Done
Inline Actions

When you'll have your fixture working as you want.
Define said lister fixture in swh/lister/launchpad/tests/conftest.py (or even just before the test that uses it, i would not mind).

@pytest.fixture
def lister_launchpad(swh_listers):
    with patch('launchpadlib.launchpad.Launchpad', MagicMock()):
        lister = swh_listers['launchpad']
    ... 
    # define how the mock should behave when calling given methods...
    # what you actually defined below in the beginning of the tests.
    ...
    lister.launchpad.git_repositories.getRepositories = MagicMock(
        side_effect=[mock_lp_response(i) for i in range(3)])
    ...

    return lister

And then instead of using swh_listers in your test definition, use the fixture define here:

def test_launchpad_lister(lister_launchpad):
    ...

That way, when reading the test, we can concentrate on checking the expected behavior of the lister.
And not the setup of the fixture (for that we will read the fixture definition ;)

ardumont: When you'll have your fixture working as you want. Define said lister fixture in…
legauAuthorUnsubmitted
Done
Inline Actions

That was my first thought but swh_listers instantiates all the listers at once so in LaunchpadLister.init login to launchpad is already called when you want to retrieve swh_listers['launchpad']

legau: That was my first thought but swh_listers instantiates all the listers at once so in…
ardumontUnsubmitted
Not Done
Inline Actions

That should not be an issue. Have you tried?

ardumont: That should not be an issue. Have you tried?
ardumontUnsubmitted
Not Done
Inline Actions

And If that becomes an issue nonetheless, do not hesitate to split the existing fixtures to be able to reuse some part within your own (as long as other tests are still working, it's fine ;):

Something like, in the main conftest:

@pytest.fixture
def lister_db_url(postgresql_proc):
    db_url = 'postgresql://{user}@{host}:{port}/{dbname}'.format(
        host=postgresql_proc.host,
        port=postgresql_proc.port,
        user='postgres',
        dbname='tests')
    logger.debug('lister db_url: %s', db_url)
    return db_url


@pytest.fixture
def swh_listers(request, lister_db_url, swh_scheduler):
    listers = {}

    # Prepare schema for all listers
    for lister_name in SUPPORTED_LISTERS:
        lister = get_lister(lister_name, db_url=lister_db_url)
        lister.scheduler = swh_scheduler  # inject scheduler fixture
        listers[lister_name] = lister
    initialize(create_engine(lister_db_url), drop_tables=True)

    # Add the load-archive-files expected by some listers (gnu, cran, ...)
    swh_scheduler.create_task_type({
        'type': 'load-archive-files',
        'description': 'Load archive files.',
        'backend_name': 'swh.loader.package.tasks.LoadArchive',
        'default_interval': '1 day',
    })

    return listers

In launchpadlister one:

@pytest.fixture
def launchpad_lister(request, lister_db_url, swh_scheduler):
    # all the necessary cogs for mocking launchpad...
    # ...
    with patch('launchpadlib.launchpad.Launchpad', MagicMock()):
        lister = get_lister('launchpad', db_url=lister_db_url)
        lister.scheduler = swh_scheduler  # inject scheduler fixture
        initialize(create_engine(lister_db_url), drop_tables=True)
    return lister
ardumont: And If that becomes an issue nonetheless, do not hesitate to split the existing fixtures to be…
legauAuthorUnsubmitted
Done
Inline Actions

Indeed it was an issue because when you use swh_listers fixture it is called before any patch so swh_listers['launchpad'].launchpad is already instanciated with a real Launchpad instance.
I'm using what you wrote it works great

legau: Indeed it was an issue because when you use swh_listers fixture it is called before any patch…
initialize(create_engine(db_url), drop_tables=True) initialize(create_engine(lister_db_url), drop_tables=True)
# Add the load-archive-files expected by some listers (gnu, cran, ...) # Add the load-archive-files expected by some listers (gnu, cran, ...)
swh_scheduler.create_task_type({ swh_scheduler.create_task_type({
'type': 'load-archive-files', 'type': 'load-archive-files',
'description': 'Load archive files.', 'description': 'Load archive files.',
'backend_name': 'swh.loader.package.tasks.LoadArchive', 'backend_name': 'swh.loader.package.tasks.LoadArchive',
'default_interval': '1 day', 'default_interval': '1 day',
}) })
return listers return listers
About · Developer information · Legal