File: test_boothook.py

package info (click to toggle)
cloud-init 25.3-2
  • links: PTS, VCS
  • area: main
  • in suites: forky, sid
  • size: 12,412 kB
  • sloc: python: 135,894; sh: 3,883; makefile: 141; javascript: 30; xml: 22
file content (64 lines) | stat: -rw-r--r-- 2,084 bytes parent folder | download | duplicates (3)
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
# This file is part of cloud-init. See LICENSE file for license information.
import re

import pytest

from cloudinit import features
from tests.integration_tests.instances import IntegrationInstance
from tests.integration_tests.util import (
    network_wait_logged,
    verify_clean_boot,
    verify_clean_log,
)

USER_DATA = """\
## template: jinja
#cloud-boothook
#!/bin/sh
# Error below will generate stderr
BOOTHOOK/0
echo BOOTHOOKstdout
echo "BOOTHOOK: {{ v1.instance_id }}: is called every boot." >> /boothook.txt
"""


@pytest.mark.user_data(USER_DATA)
class TestBoothook:
    def test_boothook_header_runs_part_per_instance(
        self,
        class_client: IntegrationInstance,
    ):
        """Test boothook handling creates a script that runs per-boot.
        Streams stderr and stdout are directed to
        /var/log/cloud-init-output.log.
        """
        client = class_client
        instance_id = client.instance.execute("cloud-init query instance-id")
        RE_BOOTHOOK = f"BOOTHOOK: {instance_id}: is called every boot"
        log = client.read_from_file("/var/log/cloud-init.log")
        verify_clean_log(log)
        verify_clean_boot(client)
        output = client.read_from_file("/boothook.txt")
        assert 1 == len(re.findall(RE_BOOTHOOK, output))
        client.restart()
        output = client.read_from_file("/boothook.txt")
        assert 2 == len(re.findall(RE_BOOTHOOK, output))
        output_log = client.read_from_file("/var/log/cloud-init-output.log")
        expected_msgs = [
            "BOOTHOOKstdout",
            "boothooks/part-001: 3: BOOTHOOK/0: not found",
        ]
        for msg in expected_msgs:
            assert msg in output_log

    def test_boothook_waits_for_network(
        self, class_client: IntegrationInstance
    ):
        """Test boothook handling waits for network before running."""
        client = class_client
        assert (
            network_wait_logged(
                client.read_from_file("/var/log/cloud-init.log")
            )
            == features.MANUAL_NETWORK_WAIT
        )