test(services, system): untie dkim tests from rest
parent
7038d69069
commit
9f04729296
|
@ -6,6 +6,7 @@ import json
|
||||||
import os
|
import os
|
||||||
import subprocess
|
import subprocess
|
||||||
import portalocker
|
import portalocker
|
||||||
|
import typing
|
||||||
|
|
||||||
|
|
||||||
USERDATA_FILE = "/etc/nixos/userdata/userdata.json"
|
USERDATA_FILE = "/etc/nixos/userdata/userdata.json"
|
||||||
|
@ -166,9 +167,10 @@ def parse_date(date_str: str) -> datetime.datetime:
|
||||||
raise ValueError("Invalid date string")
|
raise ValueError("Invalid date string")
|
||||||
|
|
||||||
|
|
||||||
def get_dkim_key(domain, parse=True):
|
def get_dkim_key(domain: str, parse: bool = True) -> typing.Optional[str]:
|
||||||
"""Get DKIM key from /var/dkim/<domain>.selector.txt"""
|
"""Get DKIM key from /var/dkim/<domain>.selector.txt"""
|
||||||
if os.path.exists("/var/dkim/" + domain + ".selector.txt"):
|
if os.path.exists("/var/dkim/" + domain + ".selector.txt"):
|
||||||
|
# Is this really neccessary to use Popen here?
|
||||||
cat_process = subprocess.Popen(
|
cat_process = subprocess.Popen(
|
||||||
["cat", "/var/dkim/" + domain + ".selector.txt"], stdout=subprocess.PIPE
|
["cat", "/var/dkim/" + domain + ".selector.txt"], stdout=subprocess.PIPE
|
||||||
)
|
)
|
||||||
|
|
|
@ -0,0 +1 @@
|
||||||
|
test-domain.tld
|
|
@ -0,0 +1,119 @@
|
||||||
|
import pytest
|
||||||
|
import typing
|
||||||
|
|
||||||
|
from os import path
|
||||||
|
from unittest.mock import DEFAULT
|
||||||
|
from tests.conftest import global_data_dir
|
||||||
|
|
||||||
|
from selfprivacy_api.utils import get_dkim_key, get_domain
|
||||||
|
import selfprivacy_api.utils as utils
|
||||||
|
|
||||||
|
###############################################################################
|
||||||
|
|
||||||
|
|
||||||
|
class ProcessMock:
|
||||||
|
"""Mock subprocess.Popen"""
|
||||||
|
|
||||||
|
def __init__(self, args, **kwargs):
|
||||||
|
self.args = args
|
||||||
|
self.kwargs = kwargs
|
||||||
|
|
||||||
|
def communicate():
|
||||||
|
return (
|
||||||
|
b'selector._domainkey\tIN\tTXT\t( "v=DKIM1; k=rsa; "\n\t "p=MIGfMA0GCSqGSIb3DQEBAQUAA4GNADCBiQKBgQDNn/IhEz1SxgHxxxI8vlPYC2dNueiLe1GC4SYz8uHimC8SDkMvAwm7rqi2SimbFgGB5nccCNOqCkrIqJTCB9vufqBnVKAjshHqpOr5hk4JJ1T/AGQKWinstmDbfTLPYTbU8ijZrwwGeqQLlnXR5nSN0GB9GazheA9zaPsT6PV+aQIDAQAB" ) ; ----- DKIM key selector for test-domain.tld\n',
|
||||||
|
None,
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
class NoFileMock(ProcessMock):
|
||||||
|
def communicate():
|
||||||
|
return (b"", None)
|
||||||
|
|
||||||
|
|
||||||
|
def _path_exists_with_masked_paths(filepath, masked_paths: typing.List[str]):
|
||||||
|
if filepath in masked_paths:
|
||||||
|
return False
|
||||||
|
else:
|
||||||
|
# this will cause the mocker to return the standard path.exists output
|
||||||
|
# see https://docs.python.org/3/library/unittest.mock.html#unittest.mock.Mock.side_effect
|
||||||
|
return DEFAULT
|
||||||
|
|
||||||
|
|
||||||
|
def path_exists_func_but_with_masked_paths(masked_paths: typing.List[str]):
|
||||||
|
"""
|
||||||
|
Sometimes we do not want to pretend that no files exist at all, but that only specific files do not exist
|
||||||
|
This provides the needed path.exists function for some arbitrary list of masked paths
|
||||||
|
"""
|
||||||
|
return lambda x: _path_exists_with_masked_paths(x, masked_paths)
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture
|
||||||
|
def mock_all_paths_exist(mocker):
|
||||||
|
mock = mocker.patch("os.path.exists", autospec=True, return_value=True)
|
||||||
|
return mock
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture
|
||||||
|
def mock_subproccess_popen_dkimfile(mocker):
|
||||||
|
mock = mocker.patch("subprocess.Popen", autospec=True, return_value=ProcessMock)
|
||||||
|
return mock
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture
|
||||||
|
def mock_subproccess_popen(mocker):
|
||||||
|
mock = mocker.patch("subprocess.Popen", autospec=True, return_value=NoFileMock)
|
||||||
|
return mock
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture
|
||||||
|
def domain_file(mocker):
|
||||||
|
# TODO: move to conftest. Challenge: it does not behave with "/" like pytest datadir does
|
||||||
|
domain_path = path.join(global_data_dir(), "domain")
|
||||||
|
mocker.patch("selfprivacy_api.utils.DOMAIN_FILE", domain_path)
|
||||||
|
return domain_path
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture
|
||||||
|
def mock_no_dkim_file(mocker):
|
||||||
|
"""
|
||||||
|
Should have domain mocks
|
||||||
|
"""
|
||||||
|
domain = utils.get_domain()
|
||||||
|
# try:
|
||||||
|
# domain = get_domain()
|
||||||
|
# except Exception as e:
|
||||||
|
# domain = ""
|
||||||
|
|
||||||
|
masked_files = ["/var/dkim/" + domain + ".selector.txt"]
|
||||||
|
mock = mocker.patch(
|
||||||
|
"os.path.exists",
|
||||||
|
side_effect=path_exists_func_but_with_masked_paths(masked_files),
|
||||||
|
)
|
||||||
|
return mock
|
||||||
|
|
||||||
|
|
||||||
|
###############################################################################
|
||||||
|
|
||||||
|
|
||||||
|
def test_get_dkim_key(
|
||||||
|
mock_subproccess_popen_dkimfile, mock_all_paths_exist, domain_file
|
||||||
|
):
|
||||||
|
"""Test DKIM key"""
|
||||||
|
dkim_key = get_dkim_key("test-domain.tld")
|
||||||
|
assert (
|
||||||
|
dkim_key
|
||||||
|
== "v=DKIM1; k=rsa; p=MIGfMA0GCSqGSIb3DQEBAQUAA4GNADCBiQKBgQDNn/IhEz1SxgHxxxI8vlPYC2dNueiLe1GC4SYz8uHimC8SDkMvAwm7rqi2SimbFgGB5nccCNOqCkrIqJTCB9vufqBnVKAjshHqpOr5hk4JJ1T/AGQKWinstmDbfTLPYTbU8ijZrwwGeqQLlnXR5nSN0GB9GazheA9zaPsT6PV+aQIDAQAB"
|
||||||
|
)
|
||||||
|
assert mock_subproccess_popen_dkimfile.call_args[0][0] == [
|
||||||
|
"cat",
|
||||||
|
"/var/dkim/test-domain.tld.selector.txt",
|
||||||
|
]
|
||||||
|
|
||||||
|
|
||||||
|
def test_no_dkim_key(
|
||||||
|
authorized_client, domain_file, mock_no_dkim_file, mock_subproccess_popen
|
||||||
|
):
|
||||||
|
"""Test no DKIM key"""
|
||||||
|
dkim_key = get_dkim_key("test-domain.tld")
|
||||||
|
assert dkim_key is None
|
||||||
|
assert mock_subproccess_popen.called == False
|
|
@ -6,6 +6,7 @@ import pytest
|
||||||
|
|
||||||
from tests.common import generate_system_query, read_json
|
from tests.common import generate_system_query, read_json
|
||||||
from tests.test_graphql.common import assert_empty
|
from tests.test_graphql.common import assert_empty
|
||||||
|
from tests.test_dkim import mock_no_dkim_file
|
||||||
|
|
||||||
|
|
||||||
@pytest.fixture
|
@pytest.fixture
|
||||||
|
@ -332,6 +333,29 @@ def test_graphql_get_domain(
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
|
def test_graphql_get_domain_no_dkim(
|
||||||
|
authorized_client,
|
||||||
|
domain_file,
|
||||||
|
mock_get_ip4,
|
||||||
|
mock_get_ip6,
|
||||||
|
mock_no_dkim_file,
|
||||||
|
turned_on,
|
||||||
|
):
|
||||||
|
"""Test no DKIM file situation gets properly handled"""
|
||||||
|
response = authorized_client.post(
|
||||||
|
"/graphql",
|
||||||
|
json={
|
||||||
|
"query": generate_system_query([API_GET_DOMAIN_INFO]),
|
||||||
|
},
|
||||||
|
)
|
||||||
|
assert response.status_code == 200
|
||||||
|
assert response.json().get("data") is not None
|
||||||
|
dns_records = response.json()["data"]["system"]["domainInfo"]["requiredDnsRecords"]
|
||||||
|
for record in dns_records:
|
||||||
|
if record["name"] == "selector._domainkey":
|
||||||
|
raise ValueError("unexpected record found:", record)
|
||||||
|
|
||||||
|
|
||||||
API_GET_TIMEZONE = """
|
API_GET_TIMEZONE = """
|
||||||
settings {
|
settings {
|
||||||
timezone
|
timezone
|
||||||
|
|
|
@ -1,102 +0,0 @@
|
||||||
import base64
|
|
||||||
import json
|
|
||||||
import pytest
|
|
||||||
|
|
||||||
from selfprivacy_api.utils import get_dkim_key
|
|
||||||
|
|
||||||
###############################################################################
|
|
||||||
|
|
||||||
|
|
||||||
class ProcessMock:
|
|
||||||
"""Mock subprocess.Popen"""
|
|
||||||
|
|
||||||
def __init__(self, args, **kwargs):
|
|
||||||
self.args = args
|
|
||||||
self.kwargs = kwargs
|
|
||||||
|
|
||||||
def communicate():
|
|
||||||
return (
|
|
||||||
b'selector._domainkey\tIN\tTXT\t( "v=DKIM1; k=rsa; "\n\t "p=MIGfMA0GCSqGSIb3DQEBAQUAA4GNADCBiQKBgQDNn/IhEz1SxgHxxxI8vlPYC2dNueiLe1GC4SYz8uHimC8SDkMvAwm7rqi2SimbFgGB5nccCNOqCkrIqJTCB9vufqBnVKAjshHqpOr5hk4JJ1T/AGQKWinstmDbfTLPYTbU8ijZrwwGeqQLlnXR5nSN0GB9GazheA9zaPsT6PV+aQIDAQAB" ) ; ----- DKIM key selector for example.com\n',
|
|
||||||
None,
|
|
||||||
)
|
|
||||||
|
|
||||||
|
|
||||||
class NoFileMock(ProcessMock):
|
|
||||||
def communicate():
|
|
||||||
return (b"", None)
|
|
||||||
|
|
||||||
|
|
||||||
@pytest.fixture
|
|
||||||
def mock_subproccess_popen(mocker):
|
|
||||||
mock = mocker.patch("subprocess.Popen", autospec=True, return_value=ProcessMock)
|
|
||||||
mocker.patch(
|
|
||||||
"selfprivacy_api.rest.services.get_domain",
|
|
||||||
autospec=True,
|
|
||||||
return_value="example.com",
|
|
||||||
)
|
|
||||||
mocker.patch("os.path.exists", autospec=True, return_value=True)
|
|
||||||
return mock
|
|
||||||
|
|
||||||
|
|
||||||
@pytest.fixture
|
|
||||||
def mock_no_file(mocker):
|
|
||||||
mock = mocker.patch("subprocess.Popen", autospec=True, return_value=NoFileMock)
|
|
||||||
mocker.patch(
|
|
||||||
"selfprivacy_api.rest.services.get_domain",
|
|
||||||
autospec=True,
|
|
||||||
return_value="example.com",
|
|
||||||
)
|
|
||||||
mocker.patch("os.path.exists", autospec=True, return_value=False)
|
|
||||||
return mock
|
|
||||||
|
|
||||||
|
|
||||||
###############################################################################
|
|
||||||
|
|
||||||
|
|
||||||
def test_unauthorized(client, mock_subproccess_popen):
|
|
||||||
"""Test unauthorized"""
|
|
||||||
response = client.get("/services/mailserver/dkim")
|
|
||||||
assert response.status_code == 401
|
|
||||||
|
|
||||||
|
|
||||||
def test_illegal_methods(authorized_client, mock_subproccess_popen):
|
|
||||||
response = authorized_client.post("/services/mailserver/dkim")
|
|
||||||
assert response.status_code == 405
|
|
||||||
response = authorized_client.put("/services/mailserver/dkim")
|
|
||||||
assert response.status_code == 405
|
|
||||||
response = authorized_client.delete("/services/mailserver/dkim")
|
|
||||||
assert response.status_code == 405
|
|
||||||
|
|
||||||
|
|
||||||
def test_get_dkim_key(mock_subproccess_popen):
|
|
||||||
"""Test DKIM key"""
|
|
||||||
dkim_key = get_dkim_key("example.com")
|
|
||||||
assert (
|
|
||||||
dkim_key
|
|
||||||
== "v=DKIM1; k=rsa; p=MIGfMA0GCSqGSIb3DQEBAQUAA4GNADCBiQKBgQDNn/IhEz1SxgHxxxI8vlPYC2dNueiLe1GC4SYz8uHimC8SDkMvAwm7rqi2SimbFgGB5nccCNOqCkrIqJTCB9vufqBnVKAjshHqpOr5hk4JJ1T/AGQKWinstmDbfTLPYTbU8ijZrwwGeqQLlnXR5nSN0GB9GazheA9zaPsT6PV+aQIDAQAB"
|
|
||||||
)
|
|
||||||
assert mock_subproccess_popen.call_args[0][0] == [
|
|
||||||
"cat",
|
|
||||||
"/var/dkim/example.com.selector.txt",
|
|
||||||
]
|
|
||||||
|
|
||||||
|
|
||||||
def test_dkim_key(authorized_client, mock_subproccess_popen):
|
|
||||||
"""Test old REST DKIM key endpoint"""
|
|
||||||
response = authorized_client.get("/services/mailserver/dkim")
|
|
||||||
assert response.status_code == 200
|
|
||||||
assert (
|
|
||||||
base64.b64decode(response.text)
|
|
||||||
== b'selector._domainkey\tIN\tTXT\t( "v=DKIM1; k=rsa; "\n\t "p=MIGfMA0GCSqGSIb3DQEBAQUAA4GNADCBiQKBgQDNn/IhEz1SxgHxxxI8vlPYC2dNueiLe1GC4SYz8uHimC8SDkMvAwm7rqi2SimbFgGB5nccCNOqCkrIqJTCB9vufqBnVKAjshHqpOr5hk4JJ1T/AGQKWinstmDbfTLPYTbU8ijZrwwGeqQLlnXR5nSN0GB9GazheA9zaPsT6PV+aQIDAQAB" ) ; ----- DKIM key selector for example.com\n'
|
|
||||||
)
|
|
||||||
assert mock_subproccess_popen.call_args[0][0] == [
|
|
||||||
"cat",
|
|
||||||
"/var/dkim/example.com.selector.txt",
|
|
||||||
]
|
|
||||||
|
|
||||||
|
|
||||||
def test_no_dkim_key(authorized_client, mock_no_file):
|
|
||||||
"""Test no DKIM key"""
|
|
||||||
response = authorized_client.get("/services/mailserver/dkim")
|
|
||||||
assert response.status_code == 404
|
|
||||||
assert mock_no_file.called == False
|
|
Loading…
Reference in New Issue