skip_changelog: make testinfra tests capture logs on service failure

Signed-off-by: gardar <gardar@users.noreply.github.com>
This commit is contained in:
gardar 2023-08-18 15:47:32 +00:00
parent 764f6db7bc
commit 083ff4ef36
No known key found for this signature in database
GPG key ID: 75FAE37CBA8C13C2
29 changed files with 254 additions and 93 deletions

View file

@ -34,8 +34,15 @@ def test_files(host, files):
def test_service(host):
s = host.service("alertmanager")
# assert s.is_enabled
assert s.is_running
try:
assert s.is_running
except AssertionError:
# Capture service logs
journal_output = host.run('journalctl -u alertmanager --since "1 hour ago"')
print("\n==== journalctl -u alertmanager Output ====\n")
print(journal_output)
print("\n============================================\n")
raise # Re-raise the original assertion error
@pytest.mark.parametrize("sockets", [

View file

@ -34,8 +34,15 @@ def test_files(host, files):
def test_service(host):
s = host.service("alertmanager")
# assert s.is_enabled
assert s.is_running
try:
assert s.is_running
except AssertionError:
# Capture service logs
journal_output = host.run('journalctl -u alertmanager --since "1 hour ago"')
print("\n==== journalctl -u alertmanager Output ====\n")
print(journal_output)
print("\n============================================\n")
raise # Re-raise the original assertion error
def test_socket(host):

View file

@ -22,8 +22,15 @@ def test_files(host, files):
def test_service(host):
s = host.service("alertmanager")
# assert s.is_enabled
assert s.is_running
try:
assert s.is_running
except AssertionError:
# Capture service logs
journal_output = host.run('journalctl -u alertmanager --since "1 hour ago"')
print("\n==== journalctl -u alertmanager Output ====\n")
print(journal_output)
print("\n============================================\n")
raise # Re-raise the original assertion error
def test_socket(host):

View file

@ -22,8 +22,15 @@ def test_files(host, files):
def test_service(host):
s = host.service("blackbox_exporter")
assert s.is_running
# assert s.is_enabled
try:
assert s.is_running
except AssertionError:
# Capture service logs
journal_output = host.run('journalctl -u blackbox_exporter --since "1 hour ago"')
print("\n==== journalctl -u blackbox_exporter Output ====\n")
print(journal_output)
print("\n============================================\n")
raise # Re-raise the original assertion error
def test_socket(host):

View file

@ -22,8 +22,15 @@ def test_files(host, files):
def test_service(host):
s = host.service("blackbox_exporter")
assert s.is_running
# assert s.is_enabled
try:
assert s.is_running
except AssertionError:
# Capture service logs
journal_output = host.run('journalctl -u blackbox_exporter --since "1 hour ago"')
print("\n==== journalctl -u blackbox_exporter Output ====\n")
print(journal_output)
print("\n============================================\n")
raise # Re-raise the original assertion error
def test_socket(host):

View file

@ -20,7 +20,15 @@ def test_directories(host):
def test_service(host):
s = host.service("chrony_exporter")
assert s.is_running
try:
assert s.is_running
except AssertionError:
# Capture service logs
journal_output = host.run('journalctl -u chrony_exporter --since "1 hour ago"')
print("\n==== journalctl -u chrony_exporter Output ====\n")
print(journal_output)
print("\n============================================\n")
raise # Re-raise the original assertion error
def test_protecthome_property(host):

View file

@ -53,8 +53,15 @@ def test_user(host):
def test_service(host):
s = host.service("chrony_exporter")
# assert s.is_enabled
assert s.is_running
try:
assert s.is_running
except AssertionError:
# Capture service logs
journal_output = host.run('journalctl -u chrony_exporter --since "1 hour ago"')
print("\n==== journalctl -u chrony_exporter Output ====\n")
print(journal_output)
print("\n============================================\n")
raise # Re-raise the original assertion error
def test_protecthome_property(host):

View file

@ -21,8 +21,15 @@ def test_files(host, files):
def test_service(host):
s = host.service("chrony_exporter")
# assert s.is_enabled
assert s.is_running
try:
assert s.is_running
except AssertionError:
# Capture service logs
journal_output = host.run('journalctl -u chrony_exporter --since "1 hour ago"')
print("\n==== journalctl -u chrony_exporter Output ====\n")
print(journal_output)
print("\n============================================\n")
raise # Re-raise the original assertion error
def test_protecthome_property(host):

View file

@ -10,8 +10,15 @@ testinfra_hosts = testinfra.utils.ansible_runner.AnsibleRunner(
def test_service(host):
s = host.service("mysqld_exporter")
# assert s.is_enabled
assert s.is_running
try:
assert s.is_running
except AssertionError:
# Capture service logs
journal_output = host.run('journalctl -u mysqld_exporter --since "1 hour ago"')
print("\n==== journalctl -u mysqld_exporter Output ====\n")
print(journal_output)
print("\n============================================\n")
raise # Re-raise the original assertion error
def test_socket(host):

View file

@ -53,8 +53,15 @@ def test_user(host):
def test_service(host):
s = host.service("mysqld_exporter")
# assert s.is_enabled
assert s.is_running
try:
assert s.is_running
except AssertionError:
# Capture service logs
journal_output = host.run('journalctl -u mysqld_exporter --since "1 hour ago"')
print("\n==== journalctl -u mysqld_exporter Output ====\n")
print(journal_output)
print("\n============================================\n")
raise # Re-raise the original assertion error
def test_socket(host):

View file

@ -21,8 +21,15 @@ def test_files(host, files):
def test_service(host):
s = host.service("mysqld_exporter")
# assert s.is_enabled
assert s.is_running
try:
assert s.is_running
except AssertionError:
# Capture service logs
journal_output = host.run('journalctl -u mysqld_exporter --since "1 hour ago"')
print("\n==== journalctl -u mysqld_exporter Output ====\n")
print(journal_output)
print("\n============================================\n")
raise # Re-raise the original assertion error
def test_socket(host):

View file

@ -19,8 +19,15 @@ def test_directories(host):
def test_service(host):
s = host.service("node_exporter")
# assert s.is_enabled
assert s.is_running
try:
assert s.is_running
except AssertionError:
# Capture service logs
journal_output = host.run('journalctl -u node_exporter --since "1 hour ago"')
print("\n==== journalctl -u node_exporter Output ====\n")
print(journal_output)
print("\n============================================\n")
raise # Re-raise the original assertion error
def test_protecthome_property(host):

View file

@ -53,8 +53,15 @@ def test_user(host):
def test_service(host):
s = host.service("node_exporter")
# assert s.is_enabled
assert s.is_running
try:
assert s.is_running
except AssertionError:
# Capture service logs
journal_output = host.run('journalctl -u node_exporter --since "1 hour ago"')
print("\n==== journalctl -u node_exporter Output ====\n")
print(journal_output)
print("\n============================================\n")
raise # Re-raise the original assertion error
def test_protecthome_property(host):

View file

@ -21,8 +21,15 @@ def test_files(host, files):
def test_service(host):
s = host.service("node_exporter")
# assert s.is_enabled
assert s.is_running
try:
assert s.is_running
except AssertionError:
# Capture service logs
journal_output = host.run('journalctl -u node_exporter --since "1 hour ago"')
print("\n==== journalctl -u node_exporter Output ====\n")
print(journal_output)
print("\n============================================\n")
raise # Re-raise the original assertion error
def test_protecthome_property(host):

View file

@ -52,8 +52,15 @@ def test_file_contents(host, file, content):
def test_service(host):
s = host.service("prometheus")
# assert s.is_enabled
assert s.is_running
try:
assert s.is_running
except AssertionError:
# Capture service logs
journal_output = host.run('journalctl -u prometheus --since "1 hour ago"')
print("\n==== journalctl -u prometheus Output ====\n")
print(journal_output)
print("\n============================================\n")
raise # Re-raise the original assertion error
def test_socket(host):

View file

@ -60,8 +60,15 @@ def test_user(host):
def test_service(host):
s = host.service("prometheus")
# assert s.is_enabled
assert s.is_running
try:
assert s.is_running
except AssertionError:
# Capture service logs
journal_output = host.run('journalctl -u prometheus --since "1 hour ago"')
print("\n==== journalctl -u prometheus Output ====\n")
print(journal_output)
print("\n============================================\n")
raise # Re-raise the original assertion error
def test_socket(host):

View file

@ -22,8 +22,15 @@ def test_files(host, files):
def test_service(host):
s = host.service("prometheus")
# assert s.is_enabled
assert s.is_running
try:
assert s.is_running
except AssertionError:
# Capture service logs
journal_output = host.run('journalctl -u prometheus --since "1 hour ago"')
print("\n==== journalctl -u prometheus Output ====\n")
print(journal_output)
print("\n============================================\n")
raise # Re-raise the original assertion error
def test_socket(host):

View file

@ -20,8 +20,15 @@ def test_directories(host):
def test_service(host):
s = host.service("pushgateway")
# assert s.is_enabled
assert s.is_running
try:
assert s.is_running
except AssertionError:
# Capture service logs
journal_output = host.run('journalctl -u pushgateway --since "1 hour ago"')
print("\n==== journalctl -u pushgateway Output ====\n")
print(journal_output)
print("\n============================================\n")
raise # Re-raise the original assertion error
def test_protecthome_property(host):

View file

@ -53,8 +53,15 @@ def test_user(host):
def test_service(host):
s = host.service("pushgateway")
# assert s.is_enabled
assert s.is_running
try:
assert s.is_running
except AssertionError:
# Capture service logs
journal_output = host.run('journalctl -u pushgateway --since "1 hour ago"')
print("\n==== journalctl -u pushgateway Output ====\n")
print(journal_output)
print("\n============================================\n")
raise # Re-raise the original assertion error
def test_protecthome_property(host):

View file

@ -21,8 +21,15 @@ def test_files(host, files):
def test_service(host):
s = host.service("pushgateway")
# assert s.is_enabled
assert s.is_running
try:
assert s.is_running
except AssertionError:
# Capture service logs
journal_output = host.run('journalctl -u pushgateway --since "1 hour ago"')
print("\n==== journalctl -u pushgateway Output ====\n")
print(journal_output)
print("\n============================================\n")
raise # Re-raise the original assertion error
def test_protecthome_property(host):

View file

@ -20,8 +20,15 @@ def test_directories(host):
def test_service(host):
s = host.service("smokeping_prober")
# assert s.is_enabled
assert s.is_running
try:
assert s.is_running
except AssertionError:
# Capture service logs
journal_output = host.run('journalctl -u smokeping_prober --since "1 hour ago"')
print("\n==== journalctl -u smokeping_prober Output ====\n")
print(journal_output)
print("\n============================================\n")
raise # Re-raise the original assertion error
def test_protecthome_property(host):

View file

@ -53,8 +53,15 @@ def test_user(host):
def test_service(host):
s = host.service("smokeping_prober")
# assert s.is_enabled
assert s.is_running
try:
assert s.is_running
except AssertionError:
# Capture service logs
journal_output = host.run('journalctl -u smokeping_prober --since "1 hour ago"')
print("\n==== journalctl -u smokeping_prober Output ====\n")
print(journal_output)
print("\n============================================\n")
raise # Re-raise the original assertion error
def test_protecthome_property(host):

View file

@ -1,36 +0,0 @@
from __future__ import (absolute_import, division, print_function)
__metaclass__ = type
import os
import testinfra.utils.ansible_runner
import pytest
testinfra_hosts = testinfra.utils.ansible_runner.AnsibleRunner(
os.environ['MOLECULE_INVENTORY_FILE']).get_hosts('all')
@pytest.mark.parametrize("files", [
"/etc/systemd/system/smokeping_prober.service",
"/usr/local/bin/smokeping_prober"
])
def test_files(host, files):
f = host.file(files)
assert f.exists
assert f.is_file
def test_service(host):
s = host.service("smokeping_prober")
# assert s.is_enabled
assert s.is_running
def test_protecthome_property(host):
s = host.service("smokeping_prober")
p = s.systemd_properties
assert p.get("ProtectHome") == "yes"
def test_socket(host):
s = host.socket("tcp://0.0.0.0:9374")
assert s.is_listening

View file

@ -21,8 +21,15 @@ def test_files(host, files):
def test_service(host):
s = host.service("smokeping_prober")
# assert s.is_enabled
assert s.is_running
try:
assert s.is_running
except AssertionError:
# Capture service logs
journal_output = host.run('journalctl -u smokeping_prober --since "1 hour ago"')
print("\n==== journalctl -u smokeping_prober Output ====\n")
print(journal_output)
print("\n============================================\n")
raise # Re-raise the original assertion error
def test_socket(host):

View file

@ -22,8 +22,15 @@ def test_files(host, files):
def test_service(host):
s = host.service("snmp_exporter")
# assert s.is_enabled
assert s.is_running
try:
assert s.is_running
except AssertionError:
# Capture service logs
journal_output = host.run('journalctl -u snmp_exporter --since "1 hour ago"')
print("\n==== journalctl -u snmp_exporter Output ====\n")
print(journal_output)
print("\n============================================\n")
raise # Re-raise the original assertion error
def test_socket(host):

View file

@ -22,8 +22,15 @@ def test_files(host, files):
def test_service(host):
s = host.service("snmp_exporter")
# assert s.is_enabled
assert s.is_running
try:
assert s.is_running
except AssertionError:
# Capture service logs
journal_output = host.run('journalctl -u snmp_exporter --since "1 hour ago"')
print("\n==== journalctl -u snmp_exporter Output ====\n")
print(journal_output)
print("\n============================================\n")
raise # Re-raise the original assertion error
def test_socket(host):

View file

@ -10,8 +10,15 @@ testinfra_hosts = testinfra.utils.ansible_runner.AnsibleRunner(
def test_service(host):
s = host.service("systemd_exporter")
# assert s.is_enabled
assert s.is_running
try:
assert s.is_running
except AssertionError:
# Capture service logs
journal_output = host.run('journalctl -u systemd_exporter --since "1 hour ago"')
print("\n==== journalctl -u systemd_exporter Output ====\n")
print(journal_output)
print("\n============================================\n")
raise # Re-raise the original assertion error
def test_socket(host):

View file

@ -41,10 +41,17 @@ def test_user(host):
assert host.user("systemd-exporter").home == "/"
# def test_service(host):
# s = host.service("systemd_exporter")
# # assert s.is_enabled
# assert s.is_running
def test_service(host):
s = host.service("systemd_exporter")
try:
assert s.is_running
except AssertionError:
# Capture service logs
journal_output = host.run('journalctl -u systemd_exporter --since "1 hour ago"')
print("\n==== journalctl -u systemd_exporter Output ====\n")
print(journal_output)
print("\n============================================\n")
raise # Re-raise the original assertion error
def test_socket(host):

View file

@ -21,8 +21,15 @@ def test_files(host, files):
def test_service(host):
s = host.service("systemd_exporter")
# assert s.is_enabled
assert s.is_running
try:
assert s.is_running
except AssertionError:
# Capture service logs
journal_output = host.run('journalctl -u systemd_exporter --since "1 hour ago"')
print("\n==== journalctl -u systemd_exporter Output ====\n")
print(journal_output)
print("\n============================================\n")
raise # Re-raise the original assertion error
def test_socket(host):