foremanctl/tests/foreman_test.py at 9cf9adea4f5c72574f0da80fdd72d945abafabac · theforeman/foremanctl · GitHub

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
import json

import pytest

FOREMAN_HOST = 'localhost'
FOREMAN_PORT = 3000

RECURRING_INSTANCES = [
    "hourly",
    "daily",
    "weekly",
    "monthly",
]

@pytest.fixture(scope="module")
def foreman_status_curl(server, server_fqdn):
    return server.run(f"curl --header 'X-FORWARDED-PROTO: https' --silent --write-out '%{{stderr}}%{{http_code}}' http://{server_fqdn}:{FOREMAN_PORT}/api/v2/ping")

@pytest.fixture(scope="module")
def foreman_status(foreman_status_curl):
    return json.loads(foreman_status_curl.stdout)


def test_foreman_service(server):
    foreman = server.service("foreman")
    assert foreman.is_running


def test_foreman_port(server):
    foreman = server.addr(FOREMAN_HOST)
    assert foreman.port(FOREMAN_PORT).is_reachable


def test_foreman_status(foreman_status_curl):
    assert foreman_status_curl.succeeded
    assert foreman_status_curl.stderr == '200'


def test_foreman_status_database(foreman_status):
    assert foreman_status['results']['foreman']['database']['active']


def test_foreman_status_cache(foreman_status):
    assert foreman_status['results']['foreman']['cache']['servers']
    assert foreman_status['results']['foreman']['cache']['servers'][0]['status'] == 'ok'


@pytest.mark.parametrize("katello_service", ['candlepin', 'candlepin_auth', 'candlepin_events', 'foreman_tasks', 'katello_events', 'pulp3', 'pulp3_content'])
def test_katello_services_status(foreman_status, katello_service):
    assert foreman_status['results']['katello']['services'][katello_service]['status'] == 'ok'


@pytest.mark.parametrize("dynflow_instance", ['orchestrator', 'worker', 'worker-hosts-queue'])
def test_foreman_dynflow_container_instances(server, dynflow_instance):
    file = server.file(f"/etc/containers/systemd/dynflow-sidekiq@{dynflow_instance}.container")
    assert file.exists
    assert file.is_symlink


@pytest.mark.parametrize("dynflow_instance", ['orchestrator', 'worker', 'worker-hosts-queue'])
def test_foreman_dynflow_service_instances(server, dynflow_instance):
    service = server.service(f"dynflow-sidekiq@{dynflow_instance}")
    assert service.is_running


@pytest.mark.parametrize("instance", RECURRING_INSTANCES)
def test_foreman_recurring_timers_enabled_and_running(server, instance):
    timer = server.service(f"foreman-recurring@{instance}.timer")
    assert timer.is_enabled
    assert timer.is_running


@pytest.mark.parametrize("instance", RECURRING_INSTANCES)
def test_foreman_recurring_services_exist(server, instance):
    service = server.service(f"foreman-recurring@{instance}.service")
    assert service.exists

def test_foreman_delivery_method_setting(foremanapi):
    delivery_method_setting = foremanapi.list('settings', search='name=delivery_method')
    assert delivery_method_setting[0]['value'] == 'smtp'


def test_foreman_host_injection(server):
    cmd = server.run(f"curl --header 'X-FORWARDED-PROTO: https' --silent --write-out '%{{stderr}}%{{http_code}}' --resolve evil.hackers.test:{FOREMAN_PORT}:127.0.0.1 http://evil.hackers.test:{FOREMAN_PORT}/api/v2/ping")
    assert cmd.succeeded
    assert cmd.stderr == '403'