Add initial tests to validate foremanctl

[Gauravtalreja1]

Problem Statement

Missing coverage for new installer or foremanctl

Solution

Adding initial tests to validate foremanctl

Related Issues

[sourcery-ai]

New security issues found

[sourcery-ai]

Hey there - I've reviewed your changes - here's some feedback:

Blocking issues:

• Avoiding SQL string concatenation: untrusted input concatenated with raw SQL query can result in SQL Injection. In order to execute raw query safely, prepared statement should be used. SQLAlchemy provides TextualSQL to easily used prepared statement with named parameters. For complex SQL composition, use SQL Expression Language or Schema Definition Language. In most cases, SQLAlchemy ORM will be a better option. (link)

General comments:

• In test_new_installer.py, the SATELLITE_SERVICES list contains 'candlein' which looks like a typo for 'candlepin' and will cause the service check to behave incorrectly.
• The install_satellite_foremanctl firewalld setup uses which firewall-cmd || dnf -y install firewalld && systemctl enable --now firewalld, which due to operator precedence will skip systemctl enable --now firewalld when firewall-cmd already exists; consider splitting the checks or adding parentheses to ensure the service is always enabled when present.
• In align_to_satellite, the new_installer branch hardcodes server.version.source back to 'internal' instead of restoring the original value, which can leak config changes between different test runs; consider capturing the previous setting and restoring it in the yield teardown.

Prompt for AI Agents

Please address the comments from this code review:

## Overall Comments
- In `test_new_installer.py`, the `SATELLITE_SERVICES` list contains `'candlein'` which looks like a typo for `'candlepin'` and will cause the service check to behave incorrectly.
- The `install_satellite_foremanctl` firewalld setup uses `which firewall-cmd || dnf -y install firewalld && systemctl enable --now firewalld`, which due to operator precedence will skip `systemctl enable --now firewalld` when `firewall-cmd` already exists; consider splitting the checks or adding parentheses to ensure the service is always enabled when present.
- In `align_to_satellite`, the `new_installer` branch hardcodes `server.version.source` back to `'internal'` instead of restoring the original value, which can leak config changes between different test runs; consider capturing the previous setting and restoring it in the `yield` teardown.

## Individual Comments

### Comment 1
<location> `robottelo/hosts.py:1580-1585` </location>
<code_context>
                 satellite_repo=settings.repos.satellite_repo,
                 satmaintenance_repo=settings.repos.satmaintenance_repo,
             )
+        elif settings.server.version.source == 'upstream':
+            self.create_custom_repos(
+                foreman='https://yum.theforeman.org/nightly/el9/x86_64/',
+                foreman_plugins='http://yum.theforeman.org/plugins/nightly/el9/x86_64/',
+                katello='https://yum.theforeman.org/katello/nightly/katello/el9/x86_64/',
+            )
</code_context>

<issue_to_address>
**🚨 suggestion (security):** Use HTTPS for the `foreman_plugins` repo URL instead of HTTP.

The `foreman_plugins` repo is the only one using plain HTTP. If HTTPS is available for this endpoint, please switch to it to avoid downgrade/MiTM risk and keep the repo definitions consistent.

```suggestion
        elif settings.server.version.source == 'upstream':
            self.create_custom_repos(
                foreman='https://yum.theforeman.org/nightly/el9/x86_64/',
                foreman_plugins='https://yum.theforeman.org/plugins/nightly/el9/x86_64/',
                katello='https://yum.theforeman.org/katello/nightly/katello/el9/x86_64/',
            )
```
</issue_to_address>

### Comment 2
<location> `robottelo/hosts.py:1984-1986` </location>
<code_context>
+            self.connect()
+            assert self.is_fips_enabled()
+
+        # Configure Satellite firewall to open communication
+        assert (
+            self.execute(
+                'which firewall-cmd || dnf -y install firewalld && systemctl enable --now firewalld'
+            ).status
</code_context>

<issue_to_address>
**issue (bug_risk):** Check and assert the result of the `firewall-cmd` invocation to catch failures.

This `firewall-cmd --permanent --add-service RH-Satellite-6 && firewall-cmd --reload` sequence doesn’t have its exit status checked, unlike the firewalld install/enable step. If it fails (e.g., missing service definition or misconfigured firewalld), the deploy may appear successful but leave an unusable firewall configuration. Please assert the command’s status (or at least log and fail) so these errors are detected early.
</issue_to_address>

### Comment 3
<location> `robottelo/hosts.py:1995-1997` </location>
<code_context>
+            'firewall-cmd --permanent --add-service RH-Satellite-6 && firewall-cmd --reload'
+        )
+        # Install Satellite and return result
+        return self.execute(
+            f'foremanctl deploy --foreman-initial-admin-username {settings.server.admin_username} --foreman-initial-admin-password {settings.server.admin_password}',
+            timeout='30m',
+        )
+
</code_context>

<issue_to_address>
**🚨 issue (security):** Quote admin username/password in the `foremanctl` command to avoid shell parsing issues.

Because these values are interpolated directly into a shell command, spaces or shell metacharacters in `settings.server.admin_username` / `settings.server.admin_password` can break the command and may allow command injection if they’re not fully trusted. Please quote them (and escape embedded quotes) or use a helper that avoids invoking a shell altogether.
</issue_to_address>

### Comment 4
<location> `pytest_fixtures/core/xdist.py:16-20` </location>
<code_context>
 def align_to_satellite(request, worker_id, satellite_factory):
     """Attempt to align a Satellite to the current xdist worker"""
-    if 'build_sanity' in request.config.option.markexpr:
+    if 'new_installer' in request.config.option.markexpr:
+        settings.set('server.hostname', None)
+        settings.set('server.version.source', 'upstream')
+        yield
+        settings.set('server.version.source', 'internal')
+    elif 'build_sanity' in request.config.option.markexpr:
         settings.set("server.hostname", None)
</code_context>

<issue_to_address>
**issue (bug_risk):** Restore `server.hostname` (and possibly original `server.version.source`) after `new_installer` runs.

In the `new_installer` branch you set `server.hostname` to `None` and `server.version.source` to `'upstream'`, but only reset `server.version.source` on teardown. This can leak state into other tests and also assumes `'internal'` was the original value. Please capture the previous values before overriding and restore them after `yield`, or at minimum reset `server.hostname` as well.
</issue_to_address>

### Comment 5
<location> `tests/foreman/installer/test_new_installer.py:34` </location>
<code_context>
+    'pulp-api',
+    'pulp-content',
+    'pulp-worker@*',
+    'candlein',
+]
+
</code_context>

<issue_to_address>
**issue (bug_risk):** Service name `candlein` is likely a typo and will make this test fail even when the system is healthy.

This should probably be `candlepin` (or `candlepin.service`) to match the actual Satellite unit name and the earlier references to candlepin logs. Using `candlein` will make `systemctl is-active` always fail here, so please verify the correct unit name and update it so the health check reflects the real service state.
</issue_to_address>

### Comment 6
<location> `tests/foreman/installer/test_new_installer.py:63` </location>
<code_context>
+        deploy_network_type=settings.server.network_type,
+        host_class=Satellite,
+    ) as sat:
+        sat.install_satellite_foremanctl(
+            enable_fapolicyd=(request.param == 'fapolicyd'), enable_fips=(request.param == 'fips')
+        )
</code_context>

<issue_to_address>
**suggestion (testing):** The fixture does not assert that `foremanctl deploy` succeeded before yielding the Satellite instance.

`install_satellite_foremanctl` returns the `foremanctl deploy` result, but the fixture ignores it and yields `sat` either way, so tests can run even if deployment failed. Please capture the return value and assert success before yielding, e.g.:

```python
result = sat.install_satellite_foremanctl(
    enable_fapolicyd=(request.param == 'fapolicyd'),
    enable_fips=(request.param == 'fips'),
)
assert result.status == 0, result.stdout
```

This makes failures surface at fixture setup instead of later in unrelated tests.

Suggested implementation:

```python
        deploy_network_type=settings.server.network_type,
        host_class=Satellite,
    ) as sat:
        result = sat.install_satellite_foremanctl(
            enable_fapolicyd=(request.param == 'fapolicyd'),
            enable_fips=(request.param == 'fips'),
        )
        assert result.status == 0, result.stdout

    assert not result.stdout

```

From the snippet you shared, the fixture body appears incomplete or mis-indented (e.g. `satellite` vs `sat`, `result` used before assignment, and no `yield`). To fully implement the intended behavior, you should also:

1. Ensure the fixture actually yields the `Satellite` instance, e.g., `yield sat`, and move any post-test log checks (like greps in `/var/log/httpd/*`) after the `yield`.
2. Make the variable name consistent (`sat` vs `satellite`) throughout the fixture.
3. Initialize `result` appropriately before any `assert not result.stdout` statements that are not intended to test the deploy result.

Those changes will depend on the full content of `module_sat_ready_rhel`, so they should be adjusted in the rest of the file accordingly.
</issue_to_address>

### Comment 7
<location> `tests/foreman/installer/test_new_installer.py:100-103` </location>
<code_context>
+
+    :expectedresults: All Satellite services are active
+    """
+    for service in SATELLITE_SERVICES:
+        is_active = module_sat_ready_rhel.execute(f'systemctl is-active {service}')
+        status = module_sat_ready_rhel.execute(f'systemctl status {service}')
+        assert is_active.status == 0, status.stdout
+
+
</code_context>

<issue_to_address>
**issue:** Service checks are very broad and may not correctly handle templated units like `pulp-worker@*`.

Because `SATELLITE_SERVICES` includes `pulp-worker@*`, `systemctl is-active pulp-worker@*` is unlikely to work as intended: systemd expects a concrete instance name (e.g. `pulp-worker@1.service`) and won’t expand the wildcard. That can make this assertion either always fail or never actually validate the running worker instances.

You may want to handle templated units separately, e.g. by listing matching units (`systemctl list-units 'pulp-worker@*' --no-legend`) and checking each is active, or by resolving the concrete instance names in a setup step and parametrizing over those.
</issue_to_address>

### Comment 8
<location> `robottelo/hosts.py:1995-1998` </location>
<code_context>
        return self.execute(
            f'foremanctl deploy --foreman-initial-admin-username {settings.server.admin_username} --foreman-initial-admin-password {settings.server.admin_password}',
            timeout='30m',
        )
</code_context>

<issue_to_address>
**security (python.sqlalchemy.security.sqlalchemy-execute-raw-query):** Avoiding SQL string concatenation: untrusted input concatenated with raw SQL query can result in SQL Injection. In order to execute raw query safely, prepared statement should be used. SQLAlchemy provides TextualSQL to easily used prepared statement with named parameters. For complex SQL composition, use SQL Expression Language or Schema Definition Language. In most cases, SQLAlchemy ORM will be a better option.

*Source: opengrep*
</issue_to_address>

### Comment 9
<location> `tests/foreman/installer/test_new_installer.py:100-103` </location>
<code_context>

</code_context>

<issue_to_address>
**issue (code-quality):** Avoid loops in tests. ([`no-loop-in-tests`](https://docs.sourcery.ai/Reference/Rules-and-In-Line-Suggestions/Python/Default-Rules/no-loop-in-tests))

<details><summary>Explanation</summary>Avoid complex code, like loops, in test functions.

Google's software engineering guidelines says:
"Clear tests are trivially correct upon inspection"
To reach that avoid complex code in tests:
* loops
* conditionals

Some ways to fix this:

* Use parametrized tests to get rid of the loop.
* Move the complex logic into helpers.
* Move the complex part into pytest fixtures.

> Complexity is most often introduced in the form of logic. Logic is defined via the imperative parts of programming languages such as operators, loops, and conditionals. When a piece of code contains logic, you need to do a bit of mental computation to determine its result instead of just reading it off of the screen. It doesn't take much logic to make a test more difficult to reason about.

Software Engineering at Google / [Don't Put Logic in Tests](https://abseil.io/resources/swe-book/html/ch12.html#donapostrophet_put_logic_in_tests)
</details>
</issue_to_address>

### Comment 10
<location> `tests/foreman/installer/test_new_installer.py:120-122` </location>
<code_context>

</code_context>

<issue_to_address>
**issue (code-quality):** Avoid loops in tests. ([`no-loop-in-tests`](https://docs.sourcery.ai/Reference/Rules-and-In-Line-Suggestions/Python/Default-Rules/no-loop-in-tests))

<details><summary>Explanation</summary>Avoid complex code, like loops, in test functions.

Google's software engineering guidelines says:
"Clear tests are trivially correct upon inspection"
To reach that avoid complex code in tests:
* loops
* conditionals

Some ways to fix this:

* Use parametrized tests to get rid of the loop.
* Move the complex logic into helpers.
* Move the complex part into pytest fixtures.

> Complexity is most often introduced in the form of logic. Logic is defined via the imperative parts of programming languages such as operators, loops, and conditionals. When a piece of code contains logic, you need to do a bit of mental computation to determine its result instead of just reading it off of the screen. It doesn't take much logic to make a test more difficult to reason about.

Software Engineering at Google / [Don't Put Logic in Tests](https://abseil.io/resources/swe-book/html/ch12.html#donapostrophet_put_logic_in_tests)
</details>
</issue_to_address>

### Comment 11
<location> `tests/foreman/installer/test_new_installer.py:121-122` </location>
<code_context>

</code_context>

<issue_to_address>
**issue (code-quality):** Avoid conditionals in tests. ([`no-conditionals-in-tests`](https://docs.sourcery.ai/Reference/Rules-and-In-Line-Suggestions/Python/Default-Rules/no-conditionals-in-tests))

<details><summary>Explanation</summary>Avoid complex code, like conditionals, in test functions.

Google's software engineering guidelines says:
"Clear tests are trivially correct upon inspection"
To reach that avoid complex code in tests:
* loops
* conditionals

Some ways to fix this:

* Use parametrized tests to get rid of the loop.
* Move the complex logic into helpers.
* Move the complex part into pytest fixtures.

> Complexity is most often introduced in the form of logic. Logic is defined via the imperative parts of programming languages such as operators, loops, and conditionals. When a piece of code contains logic, you need to do a bit of mental computation to determine its result instead of just reading it off of the screen. It doesn't take much logic to make a test more difficult to reason about.

Software Engineering at Google / [Don't Put Logic in Tests](https://abseil.io/resources/swe-book/html/ch12.html#donapostrophet_put_logic_in_tests)
</details>
</issue_to_address>

### Comment 12
<location> `pytest_fixtures/core/xdist.py:14` </location>
<code_context>
@pytest.fixture(scope="session", autouse=True)
def align_to_satellite(request, worker_id, satellite_factory):
    """Attempt to align a Satellite to the current xdist worker"""
    if 'new_installer' in request.config.option.markexpr:
        settings.set('server.hostname', None)
        settings.set('server.version.source', 'upstream')
        yield
        settings.set('server.version.source', 'internal')
    elif 'build_sanity' in request.config.option.markexpr:
        settings.set("server.hostname", None)
        yield
        # Checkout Sanity Capsule finally
        for host in [settings.capsule.hostname, settings.server.hostname]:
            if host:
                sanity_host = ContentHost.get_host_by_hostname(host)
                if settings.server.auto_checkin:
                    sanity_host.unregister()
                    Broker(hosts=[sanity_host]).checkin()
    else:
        # clear any hostname that may have been previously set
        settings.set("server.hostname", None)
        on_demand_sat = None

        worker_pos = 0 if worker_id in ["master", "local"] else int(worker_id.replace("gw", ""))

        # attempt to add potential satellites from the broker inventory file
        if settings.server.inventory_filter:
            logger.info(
                f'{worker_id=}: Attempting to add Satellite hosts using inventory filter: '
                f'{settings.server.inventory_filter}'
            )
            hosts = Satellite.get_hosts_from_inventory(filter=settings.server.inventory_filter)
            settings.server.hostnames += [host.hostname for host in hosts]

        logger.debug(
            f'{worker_id=}: {settings.server.xdist_behavior=}, '
            f'{settings.server.hostnames=}, {settings.server.auto_checkin=}'
        )
        # attempt to align a worker to a satellite
        if settings.server.xdist_behavior == 'run-on-one' and settings.server.hostnames:
            settings.set("server.hostname", settings.server.hostnames[0])
        elif settings.server.hostnames and worker_pos < len(settings.server.hostnames):
            settings.set("server.hostname", settings.server.hostnames[worker_pos])
        elif settings.server.xdist_behavior == 'balance' and settings.server.hostnames:
            settings.set("server.hostname", random.choice(settings.server.hostnames))
        # get current satellite information
        elif settings.server.xdist_behavior == 'on-demand':
            on_demand_sat = satellite_factory()
            if on_demand_sat.hostname:
                settings.set("server.hostname", on_demand_sat.hostname)
            # if no satellite was received, fallback to balance
            if not settings.server.hostname:
                logger.info(
                    f'{worker_id=}: No Satellite hostnames were available, '
                    'falling back to balance behavior'
                )
                settings.set("server.hostname", random.choice(settings.server.hostnames))
        if settings.server.hostname:
            logger.info(f'{worker_id=}: Worker was assigned hostname {settings.server.hostname}')
            configure_airgun()
            configure_nailgun()
        yield
        if on_demand_sat and settings.server.auto_checkin:
            logger.info(f'{worker_id=}: Checking in on-demand Satellite {on_demand_sat.hostname}')
            on_demand_sat.teardown()
            Broker(hosts=[on_demand_sat]).checkin()

</code_context>

<issue_to_address>
**issue (code-quality):** Low code quality found in align\_to\_satellite - 15% ([`low-code-quality`](https://docs.sourcery.ai/Reference/Default-Rules/comments/low-code-quality/))

<br/><details><summary>Explanation</summary>The quality score for this function is below the quality threshold of 25%.
This score is a combination of the method length, cognitive complexity and working memory.

How can you solve this?

It might be worth refactoring this function to make it shorter and more readable.

- Reduce the function length by extracting pieces of functionality out into
  their own functions. This is the most important thing you can do - ideally a
  function should be less than 10 lines.
- Reduce nesting, perhaps by introducing guard clauses to return early.
- Ensure that variables are tightly scoped, so that code using related concepts
  sits together within the function rather than being scattered.</details>
</issue_to_address>

---

Sourcery is free for open source - if you like our reviews please consider sharing them ✨

• X
• Mastodon
• LinkedIn
• Facebook

_{Help me be more useful! Please click 👍 or 👎 on each comment and I'll use the feedback to improve your reviews.}

[Gauravtalreja1]

trigger: test-robottelo
pytest: tests/foreman/installer/test_new_installer.py

[ogajduse]

Here are my two T2 cents.

[Gauravtalreja1]

trigger: test-robottelo
pytest: tests/foreman/installer/test_new_installer.py

[Satellite-QE]

PRT Result

Build Number: 14400
Build Status: UNSTABLE
PRT Comment: pytest tests/foreman/installer/test_new_installer.py --external-logging
Test Result : ============ 1 failed, 13 passed, 63 warnings in 1796.79s (0:29:56) ============

[shubhamsg199]

trigger: test-robottelo
pytest: tests/foreman/installer/test_new_installer.py

[Satellite-QE]

PRT Result

Build Number: 14615
Build Status: UNSTABLE
PRT Comment: pytest tests/foreman/installer/test_new_installer.py --external-logging
Test Result : ============ 2 failed, 12 passed, 70 warnings in 1991.79s (0:33:11) ============

[Gauravtalreja1]

trigger: test-robottelo
pytest: tests/foreman/installer/test_new_installer.py

[Satellite-QE]

PRT Result

Build Number: 14758
Build Status: UNSTABLE
PRT Comment: pytest tests/foreman/installer/test_new_installer.py --external-logging
Test Result : ============ 1 failed, 13 passed, 60 warnings in 1806.68s (0:30:06) ============

[Gauravtalreja1]

trigger: test-robottelo
pytest: tests/foreman/installer/test_new_installer.py

[evgeni]

Given the list here is the same as SATELLITE_SERVICES, you could do

Suggested change

	result = satellite.execute(
	r'journalctl --quiet --no-pager --boot --grep ERROR -u "dynflow-sidekiq*" -u "foreman-proxy" -u "foreman" -u "httpd" -u "postgresql" -u "pulp-api" -u "pulp-content" -u "pulp-worker*" -u "redis" -u "candlepin"'
	)
	units_string = ' -u '.join([f'"{service}"' for service in SATELLITE_SERVICES])
	result = satellite.execute(f'journalctl --quiet --no-pager --boot --grep ERROR -u {units_string}')

It makes the test re-use the list, but is not really well readable :/

(Why is this a raw string btw?)

[Gauravtalreja1]

It does not exactly follow a SATELLITE_SERVICES as dynflow-sidekiq* and pulp-worker* is not a service, and defining a separate list for this systemd units would not make any sense as it wouldn't be reusable, and I see no problem with it being in raw string

[evgeni]

SATELLITE_SERVICES contains pulp-worker@*, the list here has pulp-worker*, and I would expect both match "all pulp worker units"
Same-ish for dynflow-sidekiq* here which should match the three instances listed in SATELLITE_SERVICES

So yes, the lists are not identical, but I'd expect the result to be.

That said, this all is not blocking and I am totally OK with the current code.

[Satellite-QE]

PRT Result

Build Number: 14763
Build Status: SUCCESS
PRT Comment: pytest tests/foreman/installer/test_new_installer.py --external-logging
Test Result : ================= 14 passed, 60 warnings in 1765.34s (0:29:25) =================