/
conftest.py
95 lines (76 loc) · 2.97 KB
/
conftest.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
"""Configuration for unit tests."""
import pytest
from flexmock import flexmock
from f8a_worker.enums import EcosystemBackend
from f8a_worker.models import Base, Ecosystem, create_db_scoped_session
from f8a_worker.setup_celery import get_dispatcher_config_files
from f8a_worker.storages import AmazonS3
from selinon import Config
# To use fixtures from this file, either name them as an input argument or use 'usefixtures' marker
# https://docs.pytest.org/en/latest/fixture.html#using-fixtures-from-classes-modules-or-projects
@pytest.fixture
def rdb():
"""Provide connection to a database."""
session = create_db_scoped_session()
# TODO: we may need to run actual migrations here
# make sure all session objects from scoped_session get closed here
# otherwise drop_all() would hang indefinitely
session.close_all()
# NOTE: this also drops any data created by fixtures (e.g. builtin ecosystems),
# so if you want to use these, create them by hand before running your tests
# We can't use Base.metadata.drop_all(bind=session.bind), since they may be tables from
# e.g. bayesian server, that reference f8a_worker tables and will prevent dropping them
tables = session.bind.table_names()
for t in tables:
session.execute('drop table if exists "{t}" cascade'.format(t=t))
session.commit()
Base.metadata.create_all(bind=session.bind)
return session
@pytest.fixture
def maven(rdb):
"""Prepare database with Maven ecosystem."""
maven = Ecosystem(name='maven', backend=EcosystemBackend.maven,
fetch_url='https://repo.maven.apache.org/maven2/')
rdb.add(maven)
rdb.commit()
return maven
@pytest.fixture
def npm(rdb):
"""Prepare database with NPM ecosystem."""
npm = Ecosystem(name='npm', backend=EcosystemBackend.npm,
fetch_url='https://registry.npmjs.org/')
rdb.add(npm)
rdb.commit()
return npm
@pytest.fixture
def pypi(rdb):
"""Prepare database with Pypi ecosystem."""
pypi = Ecosystem(name='pypi', backend=EcosystemBackend.pypi,
fetch_url='https://pypi.org/pypi/')
rdb.add(pypi)
rdb.commit()
return pypi
@pytest.fixture
def nuget(rdb):
"""Prepare database with Nuget ecosystem."""
nuget = Ecosystem(name='nuget', backend=EcosystemBackend.nuget,
fetch_url='https://api.nuget.org/packages/')
rdb.add(nuget)
rdb.commit()
return nuget
@pytest.fixture
def go(rdb):
"""Prepare database with Go ecosystem."""
e = Ecosystem(name='go', backend=EcosystemBackend.go, fetch_url='')
rdb.add(e)
rdb.commit()
return e
@pytest.fixture()
def dispatcher_setup():
"""Perform environment setup for Dispatcher if needed."""
nodes_yaml, flows_yaml = get_dispatcher_config_files()
Config.set_config_yaml(nodes_yaml, flows_yaml)
@pytest.fixture()
def no_s3_connection():
"""Mock the connection to S3."""
flexmock(AmazonS3).should_receive('is_connected').and_return(True)