"""Fixtures for testing CHAPPS adapters module"""
from unittest.mock import Mock
import pytest
from pytest import fixture
import MySQLdb as dbmodule
from chapps.adapter import (
PolicyConfigAdapter,
MariaDBQuotaAdapter,
MariaDBSenderDomainAuthAdapter,
MariaDBInboundFlagsAdapter,
)
from chapps.tests.test_sqla_adapter.conftest import (
no_options_domain,
greylisting_domain,
spf_domain,
enforcing_both_domain,
_populated_database_fixture,
_populated_database_fixture_with_extras,
_populated_database_fixture_with_breakage,
)
[docs]@fixture
def mock_dbmodule(monkeypatch):
"""Patch the mariadb module's connect function with a mock"""
monkeypatch.setattr(
dbmodule, "connect", Mock(return_value="mock connection")
)
def _adapter_fixture(fixtype):
adapter = fixtype(
db_host="localhost",
db_name="chapps_test",
db_user="chapps_test",
db_pass="screwy%pass${word}",
)
return adapter
def _database_fixture(finalizing_adapter):
cur = finalizing_adapter.conn.cursor()
cur.execute("DROP DATABASE IF EXISTS chapps_test")
cur.execute("CREATE DATABASE IF NOT EXISTS chapps_test")
cur.execute("USE chapps_test")
yield cur
cur.close()
[docs]@fixture
def base_adapter_fixture():
return _adapter_fixture(PolicyConfigAdapter)
[docs]@fixture
def finalizing_pcadapter(base_adapter_fixture):
yield base_adapter_fixture
base_adapter_fixture.finalize()
[docs]@fixture
def database_fixture(finalizing_pcadapter):
yield from _database_fixture(finalizing_pcadapter)
[docs]@fixture
def mdbqadapter_fixture():
return _mdbqadapter_fixture()
def _mdbqadapter_fixture():
return _adapter_fixture(MariaDBQuotaAdapter)
[docs]@fixture
def finalizing_mdbqadapter(mdbqadapter_fixture):
yield mdbqadapter_fixture
mdbqadapter_fixture.finalize()
[docs]@fixture
def mdbsdaadapter_fixture():
return _adapter_fixture(MariaDBSenderDomainAuthAdapter)
[docs]@fixture
def mdbifadapter_fixture():
return _adapter_fixture(MariaDBInboundFlagsAdapter)
[docs]@fixture
def finalizing_mdbsdaadapter(mdbsdaadapter_fixture):
yield mdbsdaadapter_fixture
mdbsdaadapter_fixture.finalize()
[docs]@fixture
def finalizing_mdbifadapter(mdbifadapter_fixture):
yield mdbifadapter_fixture
mdbifadapter_fixture.finalize()
[docs]@fixture
def test_emails():
return [
"ccullen@easydns.com",
"somebody@chapps.io",
"nonexistent@chapps.io",
]
# def _populated_database_fixture(database_fixture):
# basic_quotas = (
# "INSERT INTO quotas ( name, quota ) VALUES "
# "('10eph', 240),"
# "('50eph', 1200),"
# "('200eph', 4800)"
# )
# test_users = [
# "BEGIN;",
# "INSERT INTO users ( name ) VALUES ( 'ccullen@easydns.com' );",
# "SELECT LAST_INSERT_ID() INTO @userid;",
# "INSERT INTO domains ( name, greylist, check_spf )"
# " VALUES ( 'chapps.io', 1, 1 );",
# "SELECT LAST_INSERT_ID() INTO @chappsid;",
# "INSERT INTO emails (name) VALUES ( 'caleb@chapps.com' );",
# "SELECT LAST_INSERT_ID() INTO @emailid;",
# (
# "INSERT INTO quota_user ( quota_id, user_id ) VALUES"
# " ( (SELECT id FROM quotas WHERE name = '10eph'), @userid );"
# ),
# (
# "INSERT INTO domain_user ( domain_id, user_id ) VALUES"
# " ( @chappsid, @userid );"
# ),
# (
# "INSERT INTO email_user ( email_id, user_id ) VALUES"
# " ( @emailid, @userid );"
# ),
# "INSERT INTO users (name) VALUES ('somebody@chapps.io');",
# "SELECT LAST_INSERT_ID() INTO @userid;",
# (
# "INSERT INTO quota_user ( quota_id, user_id ) VALUES"
# " ( (SELECT id FROM quotas WHERE name = '50eph'), @userid );"
# ),
# (
# "INSERT INTO domain_user ( domain_id, user_id ) VALUES"
# " ( @chappsid, @userid );"
# ),
# "INSERT INTO users (name) VALUES ('bigsender@chapps.io');",
# "SELECT LAST_INSERT_ID() INTO @userid;",
# (
# "INSERT INTO quota_user ( quota_id, user_id ) VALUES"
# " ( (SELECT id FROM quotas WHERE name = '200eph'), @userid );"
# ),
# (
# "INSERT INTO domain_user ( domain_id, user_id ) VALUES"
# " ( @chappsid, @userid );"
# ),
# "COMMIT;",
# ]
# count_users = "SELECT COUNT(name) FROM users;"
# cur = database_fixture
# for policy in [MariaDBQuotaAdapter, MariaDBSenderDomainAuthAdapter]:
# _adapter_fixture(policy)._initialize_tables()
# cur.execute(count_users)
# usercount = cur.fetchone()[0]
# if usercount == 0:
# cur.execute(basic_quotas)
# for stmt in test_users:
# cur.execute(stmt)
# return cur
# # yield cur
# # could drop and re-create the DB here, but we do that elsewhere
# def _populated_database_fixture_with_extras(database_fixture):
# cur = _populated_database_fixture(database_fixture)
# extra_domains = (
# "INSERT INTO domains (name, greylist, check_spf) VALUES"
# " ('easydns.com', 0, 1),"
# " ('easydns.net', 1, 0),"
# " ('easydns.org', 0, 0);"
# )
# extra_users = (
# "INSERT INTO users (name) VALUES "
# "('schmo1@chapps.io'), ('schmo2@chapps.io');"
# )
# extra_emails = (
# "INSERT INTO emails (name) VALUES ('roleaccount@chapps.com'),"
# " ('admin@chapps.com'), ('abuse@chapps.com'), ('info@chapps.com');"
# )
# extra_assoc = [
# (
# "INSERT INTO domain_user (domain_id, user_id) VALUES"
# " (2, 3), (1, 5), (2, 5), (3, 5), (4, 5);"
# ),
# ("INSERT INTO quota_user (quota_id, user_id) VALUES (1, 5);"),
# (
# "INSERT INTO email_user (email_id, user_id) VALUES"
# " (2, 2), (2, 3), (2, 4), (2, 5),"
# " (3, 3), (4, 3), (5, 3);"
# ),
# ]
# cur.execute(extra_domains)
# cur.execute(extra_users)
# cur.execute(extra_emails)
# for q in extra_assoc:
# cur.execute(q)
# return cur
# def _populated_database_fixture_with_breakage(database_fixture):
# cur = _populated_database_fixture_with_extras(database_fixture)
# breakage = (
# "DELETE FROM quota_user WHERE user_id = 1;"
# "DELETE FROM users WHERE id = 2;"
# )
# cur.execute(breakage)
# return cur
[docs]@fixture
def populated_database_fixture(database_fixture):
return _populated_database_fixture(database_fixture)
[docs]@fixture
def populated_database_fixture_with_breakage(database_fixture):
return _populated_database_fixture_with_breakage(database_fixture)
[docs]@fixture
def mock_mapping():
m = Mock(name="mapping", ident="mock_mapping_func", return_value=True)
return m