Compare commits

...

54 Commits

Author SHA1 Message Date
Bas Nijholt
920b593d5f Fix mypy error: add type annotation for proc variable 2025-12-17 00:17:20 -08:00
Bas Nijholt
27d9b08ce2 Add -f shorthand for --full in apply command 2025-12-17 00:10:16 -08:00
Bas Nijholt
700cdacb4d Add 'a' alias for apply command (cf a = cf apply) 2025-12-17 00:09:45 -08:00
Bas Nijholt
3c7a532704 Add comments explaining lazy imports for startup performance 2025-12-17 00:08:28 -08:00
Bas Nijholt
6048f37ad5 Lazy import pydantic for faster CLI startup
- Create paths.py module with lightweight path utilities (no pydantic)
- Move Config imports to TYPE_CHECKING blocks in CLI modules
- Lazy import load_config only when needed

Combined with asyncssh lazy loading, cf --help now starts in ~150ms
instead of ~500ms (70% faster).
2025-12-17 00:07:15 -08:00
Bas Nijholt
f18952633f Lazy import asyncssh for faster CLI startup
Move asyncssh import inside _run_ssh_command() so it's only loaded
when actually executing SSH commands. This cuts CLI import time
from 414ms to 200ms (52% faster).

cf --help now starts in ~260ms instead of ~500ms.
2025-12-16 23:59:37 -08:00
Bas Nijholt
437257e631 Add cf config symlink command
Creates a symlink from ~/.config/compose-farm/compose-farm.yaml to a
local config file using absolute paths (avoids broken relative symlinks).

Usage:
  cf config symlink                     # Link to ./compose-farm.yaml
  cf config symlink /path/to/config.yaml  # Link to specific file

State files are automatically stored next to the actual config (not
the symlink) since config_path.resolve() follows symlinks.
2025-12-16 23:56:58 -08:00
Bas Nijholt
c720170f26 Add --full flag to apply command
When --full is passed, apply also runs 'docker compose up' on all
services (not just missing/migrating ones) to pick up any config
changes (compose file, .env, etc).

- cf apply          # Fast: state reconciliation only
- cf apply --full   # Thorough: also refresh all running services
2025-12-16 23:54:19 -08:00
Bas Nijholt
d9c03d6509 Feature apply as the hero command in README
- Update intro and NOTE block to lead with cf apply
- Rewrite "How It Works" to show declarative workflow first
- Move apply to top of command table (bolded)
- Reorder examples to show apply as "the main command"
- Update automation bullet to highlight one-command reconciliation
2025-12-16 23:49:49 -08:00
Bas Nijholt
3b7066711f Merge pull request #12 from basnijholt/feature/orphaned-services
Add apply command and refactor CLI for clearer UX
2025-12-16 23:34:34 -08:00
Bas Nijholt
6a630c40a1 Update apply command description to include starting missing services 2025-12-16 23:32:27 -08:00
Bas Nijholt
9f9c042b66 Remove up --migrate flag in favor of apply
Simplifies CLI by having one clear reconciliation command:
- cf up <service>  = start specific services (auto-migrates if needed)
- cf apply         = full reconcile (stop orphans + migrate + start missing)

The --migrate flag was redundant with 'apply --no-orphans'.
2025-12-16 23:27:19 -08:00
github-actions[bot]
2a6d7d0b85 Update README.md 2025-12-17 07:21:38 +00:00
Bas Nijholt
6d813ccd84 Merge af9c760fb8 into affed2edcf 2025-12-17 07:21:23 +00:00
Bas Nijholt
af9c760fb8 Add missing service detection to apply command
Previously, apply only handled:
1. Stopping orphans (in state, not in config)
2. Migrating services (in state, wrong host)

Now it also handles:
3. Starting missing services (in config, not in state)

This fixes the case where a service was stopped as an orphan, then
re-added to config - apply would say "nothing to do" instead of
starting it.

Added get_services_not_in_state() to state.py and updated tests.
2025-12-16 23:21:09 -08:00
Bas Nijholt
90656b05e3 Add tests for apply command and down --orphaned flag
Tests cover:
- apply: nothing to do, dry-run preview, migrations, orphan cleanup, --no-orphans
- down --orphaned: no orphans, stops services, error on invalid combinations

Lifecycle.py coverage improved from 20% to 61%.
2025-12-16 23:15:46 -08:00
github-actions[bot]
d7a3d4e8c7 Update README.md 2025-12-17 07:10:52 +00:00
Bas Nijholt
35f0b8bf99 Merge be6b391121 into affed2edcf 2025-12-16 23:10:36 -08:00
Bas Nijholt
be6b391121 Refactor CLI commands for clearer UX
Separate "read state from reality" from "write config to reality":
- Rename `sync` to `refresh` (updates local state from running services)
- Add `apply` command (makes reality match config: migrate + stop orphans)
- Add `down --orphaned` flag (stops services removed from config)
- Modify `up --migrate` to only handle migrations (not orphans)

The new mental model:
- `refresh` = Reality → State (discover what's running)
- `apply` = Config → Reality (reconcile: migrate services + stop orphans)

Also extract private helper functions for reporting to match codebase style.
2025-12-16 23:06:42 -08:00
Bas Nijholt
7f56ba6a41 Add orphaned service detection and cleanup
When services are removed from config but still tracked in state,
`cf up --migrate` now stops them automatically. This makes the
config truly declarative - comment out a service, run migrate,
and it stops.

Changes:
- Add get_orphaned_services() to state.py for detecting orphans
- Add stop_orphaned_services() to operations.py for cleanup
- Update lifecycle.py to call stop_orphaned_services on --migrate
- Refactor _report_orphaned_services to use shared function
- Rename "missing_from_config" to "unmanaged" for clarity
- Add tests for get_orphaned_services
- Only remove from state on successful down (not on failure)
2025-12-16 22:53:26 -08:00
Bas Nijholt
4b3d7a861e Fix migration and update for services with buildable images
Use `pull --ignore-buildable` to skip images that have `build:` defined
in the compose file, preventing pull failures for locally-built images
like gitea-runner-custom. The build step then handles these images.
2025-12-16 19:42:24 -08:00
Bas Nijholt
affed2edcf Refactor operations.py into smaller helpers
- Add PreflightResult NamedTuple with .ok property
- Extract _run_compose_step to handle raw output and interrupts
- Extract _up_single_service for single-host migration logic
- Deduplicate pull/build handling in _migrate_service
- Add was_running check to only rollback if service was running
2025-12-16 17:08:25 -08:00
Bas Nijholt
34642e8b8e Rollback to old host when migration up fails
When up fails on the target host after migration has already stopped
the service on the old host, attempt to restart on the old host as a
rollback. This prevents services from being left down after a failed
migration attempt.
2025-12-16 16:59:09 -08:00
Bas Nijholt
4c8b6c5209 Add init-network hint when network is missing 2025-12-16 16:22:06 -08:00
Bas Nijholt
2b38ed28c0 Skip traefik regeneration when all services fail
Don't update traefik config if all services in the operation failed.
This prevents adding routes for services that aren't actually running.
2025-12-16 16:21:09 -08:00
Bas Nijholt
26b57895ce Clean up orphaned containers when migration up fails
If `up` fails after migration (when we've already run `down` on the
old host), run `down` on the target host to clean up any containers
that were created but couldn't start (e.g., due to missing devices).

This prevents orphaned containers from lingering on the failed host.
2025-12-16 16:16:09 -08:00
Bas Nijholt
367da13fae Fix path existence check for permission denied
Use stat instead of test -e to distinguish "no such file" from
"permission denied". If stat fails with permission denied, the
path exists (just not accessible), so report it as existing.

Fixes false "missing path" errors for directories with restricted
permissions like /mnt/data/immich/library.
2025-12-16 15:08:38 -08:00
Bas Nijholt
d6ecd42559 Consolidate service requirement checks into shared function
Move check_service_requirements() to operations.py as a public function
that verifies paths, networks, and devices exist on a target host. Both
CLI check command and pre-flight migration checks now use this shared
function, eliminating duplication. Also adds device checking to the
check command output.
2025-12-16 14:53:59 -08:00
Bas Nijholt
233c33fa52 Add device checking to pre-flight migration checks
Services with devices: mappings (e.g., /dev/dri for GPU acceleration)
now have those devices verified on the target host before migration.
This prevents the scenario where a service is stopped on the old host
but fails to start on the new host due to missing devices.

Adds parse_devices() to extract host device paths from compose files.
2025-12-16 14:35:52 -08:00
Bas Nijholt
43974c5743 Abort on first Ctrl+C during migrations
Detect when SSH subprocess is killed by signal (exit code < 0 or 255)
and treat it as an interrupt. This allows single Ctrl+C to abort the
entire operation instead of requiring two presses.
2025-12-16 14:31:33 -08:00
Bas Nijholt
cf94a62f37 docs: Clarify pull/build comments in migration 2025-12-16 14:26:48 -08:00
Bas Nijholt
81b4074827 Pre-build Dockerfile services during migration
After pulling images, also run build for services with Dockerfiles.
This ensures build-based services have their images ready before
stopping the old service, minimizing downtime.

If build fails, abort the migration and leave the service running
on the old host.

Extract _migrate_service helper to reduce function complexity.
2025-12-16 14:17:19 -08:00
Bas Nijholt
455657c8df Abort migration if pre-pull fails
If pulling images on the target host fails (e.g., rate limit),
abort the migration and leave the service running on the old host.
This prevents downtime when Docker Hub rate limits are hit.
2025-12-16 14:14:35 -08:00
Bas Nijholt
ee5a92788a Pre-pull images during migration to reduce downtime
When migrating a service to a new host, pull images on the target
host before stopping the service on the old host. This minimizes
downtime since images are cached when the up command runs.

Migration flow:
1. Pull images on new host (service still running on old)
2. Down on old host
3. Up on new host (fast, images already pulled)
2025-12-16 14:12:53 -08:00
Bas Nijholt
2ba396a419 docs: Move Compose Farm to first column in comparison table 2025-12-16 13:48:40 -08:00
Bas Nijholt
7144d58160 build: Include LICENSE file in package distribution 2025-12-16 13:37:15 -08:00
Bas Nijholt
279fa2e5ef Create LICENSE 2025-12-16 13:36:35 -08:00
Bas Nijholt
dbe0b8b597 docs: Add app.py to CLAUDE.md architecture diagram 2025-12-16 13:14:51 -08:00
Bas Nijholt
b7315d255a refactor: Split CLI into modular subpackage (#11) 2025-12-16 13:08:08 -08:00
renovate[bot]
f003d2931f ⬆️ Update actions/checkout action to v6 (#5)
Co-authored-by: renovate[bot] <29139614+renovate[bot]@users.noreply.github.com>
2025-12-16 12:19:45 -08:00
renovate[bot]
6f7c557065 ⬆️ Update actions/setup-python action to v6 (#6)
Co-authored-by: renovate[bot] <29139614+renovate[bot]@users.noreply.github.com>
2025-12-16 12:18:34 -08:00
renovate[bot]
ecb6ee46b1 ⬆️ Update astral-sh/setup-uv action to v7 (#8)
Co-authored-by: renovate[bot] <29139614+renovate[bot]@users.noreply.github.com>
2025-12-16 12:18:28 -08:00
renovate[bot]
354967010f ⬆️ Update redis Docker tag to v8 (#9)
Co-authored-by: renovate[bot] <29139614+renovate[bot]@users.noreply.github.com>
2025-12-16 12:18:22 -08:00
github-actions[bot]
57122f31a3 Update README.md 2025-12-16 20:01:03 +00:00
Bas Nijholt
cbbcec0d14 Add config subcommand for managing configuration files (#10) 2025-12-16 12:00:44 -08:00
Bas Nijholt
de38c35b8a docs: Add one-liner showing manual equivalent 2025-12-16 11:19:56 -08:00
github-actions[bot]
def996ddf4 Update README.md 2025-12-16 19:14:07 +00:00
Bas Nijholt
790e32e96b Fix test_load_config_not_found for CF_CONFIG env var 2025-12-16 11:13:44 -08:00
Bas Nijholt
fd75c4d87f Add CLI --help output to README 2025-12-16 11:12:43 -08:00
Bas Nijholt
411a99cbc4 Wait for PyPI propagation before Docker build
Also add Python 3.14 to classifiers.
2025-12-16 11:04:35 -08:00
Bas Nijholt
d2c6ab72b2 Add CF_CONFIG env var for simpler Docker workflow
Config search order is now:
1. --config CLI option
2. CF_CONFIG environment variable
3. ./compose-farm.yaml
4. ~/.config/compose-farm/compose-farm.yaml

Docker workflow simplified: mount compose_dir once, set CF_CONFIG
to config file within it. No more symlink issues or multiple mounts.
2025-12-16 10:12:55 -08:00
Bas Nijholt
3656584eda Friendly error when config path is a directory
Docker creates empty directories for missing file mounts,
causing confusing IsADirectoryError tracebacks. Now shows
a clear message explaining the likely cause.
2025-12-16 09:49:40 -08:00
Bas Nijholt
8be370098d Use env vars for docker-compose.yml mounts
- CF_CONFIG_DIR: config directory (default: ~/.config/compose-farm)
- CF_COMPOSE_DIR: compose directory (default: /opt/compose)

Mounts preserve paths so compose_dir in config works correctly.
2025-12-16 09:49:34 -08:00
Bas Nijholt
45057cb6df feat: Add docker-compose.yml for easier Docker usage
Example compose file that mounts SSH agent and config.
Users uncomment the compose_dir mount for their setup.
2025-12-16 09:40:18 -08:00
34 changed files with 3507 additions and 1365 deletions

View File

@@ -16,10 +16,10 @@ jobs:
python-version: ["3.11", "3.12", "3.13"]
steps:
- uses: actions/checkout@v4
- uses: actions/checkout@v6
- name: Install uv
uses: astral-sh/setup-uv@v6
uses: astral-sh/setup-uv@v7
- name: Set up Python ${{ matrix.python-version }}
run: uv python install ${{ matrix.python-version }}
@@ -39,10 +39,10 @@ jobs:
lint:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- uses: actions/checkout@v6
- name: Install uv
uses: astral-sh/setup-uv@v6
uses: astral-sh/setup-uv@v7
- name: Set up Python
run: uv python install 3.12

View File

@@ -25,7 +25,7 @@ jobs:
steps:
- name: Checkout
uses: actions/checkout@v4
uses: actions/checkout@v6
- name: Set up Docker Buildx
uses: docker/setup-buildx-action@v3
@@ -52,6 +52,22 @@ jobs:
fi
echo "version=$VERSION" >> $GITHUB_OUTPUT
- name: Wait for PyPI
if: steps.version.outputs.version != ''
run: |
VERSION="${{ steps.version.outputs.version }}"
echo "Waiting for compose-farm==$VERSION on PyPI..."
for i in {1..30}; do
if curl -sf "https://pypi.org/pypi/compose-farm/$VERSION/json" > /dev/null; then
echo "✓ Version $VERSION available on PyPI"
exit 0
fi
echo "Attempt $i: not yet available, waiting 10s..."
sleep 10
done
echo "✗ Timeout waiting for PyPI"
exit 1
- name: Extract metadata
id: meta
uses: docker/metadata-action@v5

View File

@@ -13,9 +13,9 @@ jobs:
permissions:
id-token: write
steps:
- uses: actions/checkout@v4
- uses: actions/checkout@v6
- name: Install uv
uses: astral-sh/setup-uv@v6
uses: astral-sh/setup-uv@v7
- name: Build
run: uv build
- name: Publish package distributions to PyPI

View File

@@ -11,16 +11,16 @@ jobs:
runs-on: ubuntu-latest
steps:
- name: Check out repository
uses: actions/checkout@v4
uses: actions/checkout@v6
with:
persist-credentials: false
fetch-depth: 0
- name: Set up Python
uses: actions/setup-python@v5
uses: actions/setup-python@v6
- name: Install uv
uses: astral-sh/setup-uv@v6
uses: astral-sh/setup-uv@v7
- name: Run markdown-code-runner
env:

View File

@@ -10,15 +10,22 @@
```
compose_farm/
├── cli.py # Typer commands (thin layer, delegates to operations)
├── config.py # Pydantic models, YAML loading
├── compose.py # Compose file parsing (.env, ports, volumes, networks)
├── console.py # Shared Rich console instances
├── executor.py # SSH/local command execution, streaming output
├── operations.py # Business logic (up, migrate, discover, preflight checks)
├── state.py # Deployment state tracking (which service on which host)
├── logs.py # Image digest snapshots (dockerfarm-log.toml)
── traefik.py # Traefik file-provider config generation from labels
├── cli/ # CLI subpackage
│ ├── __init__.py # Imports modules to trigger command registration
│ ├── app.py # Shared Typer app instance, version callback
├── common.py # Shared helpers, options, progress bar utilities
│ ├── config.py # Config subcommand (init, show, path, validate, edit)
│ ├── lifecycle.py # up, down, pull, restart, update, apply commands
│ ├── management.py # refresh, check, init-network, traefik-file commands
│ └── monitoring.py # logs, ps, stats commands
── config.py # Pydantic models, YAML loading
├── compose.py # Compose file parsing (.env, ports, volumes, networks)
├── console.py # Shared Rich console instances
├── executor.py # SSH/local command execution, streaming output
├── operations.py # Business logic (up, migrate, discover, preflight checks)
├── state.py # Deployment state tracking (which service on which host)
├── logs.py # Image digest snapshots (dockerfarm-log.toml)
└── traefik.py # Traefik file-provider config generation from labels
```
## Key Design Decisions
@@ -48,15 +55,17 @@ CLI available as `cf` or `compose-farm`.
| Command | Description |
|---------|-------------|
| `up` | Start services (`docker compose up -d`), auto-migrates if host changed. Use `--migrate` for auto-detection |
| `down` | Stop services (`docker compose down`) |
| `up` | Start services (`docker compose up -d`), auto-migrates if host changed |
| `down` | Stop services (`docker compose down`). Use `--orphaned` to stop services removed from config |
| `pull` | Pull latest images |
| `restart` | `down` + `up -d` |
| `update` | `pull` + `down` + `up -d` |
| `apply` | Make reality match config: migrate services + stop orphans. Use `--dry-run` to preview |
| `logs` | Show service logs |
| `ps` | Show status of all services |
| `stats` | Show overview (hosts, services, pending migrations; `--live` for container counts) |
| `sync` | Discover running services, update state, capture image digests |
| `refresh` | Update state from reality: discover running services, capture image digests |
| `check` | Validate config, traefik labels, mounts, networks; show host compatibility |
| `init-network` | Create Docker network on hosts with consistent subnet/gateway |
| `traefik-file` | Generate Traefik file-provider config from compose labels |
| `config` | Manage config files (init, show, path, validate, edit) |

21
LICENSE Normal file
View File

@@ -0,0 +1,21 @@
MIT License
Copyright (c) 2025 Bas Nijholt
Permission is hereby granted, free of charge, to any person obtaining a copy
of this software and associated documentation files (the "Software"), to deal
in the Software without restriction, including without limitation the rights
to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the Software is
furnished to do so, subject to the following conditions:
The above copyright notice and this permission notice shall be included in all
copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
SOFTWARE.

163
README.md
View File

@@ -10,7 +10,7 @@
A minimal CLI tool to run Docker Compose commands across multiple hosts via SSH.
> [!NOTE]
> Run `docker compose` commands across multiple hosts via SSH. One YAML maps services to hosts. Change the mapping, run `up`, and it auto-migrates. No Kubernetes, no Swarm, no magic.
> Run `docker compose` commands across multiple hosts via SSH. One YAML maps services to hosts. Run `cf apply` and reality matches your config—services start, migrate, or stop as needed. No Kubernetes, no Swarm, no magic.
<!-- START doctoc generated TOC please keep comment here to allow auto update -->
<!-- DON'T EDIT THIS SECTION, INSTEAD RE-RUN doctoc TO UPDATE -->
@@ -25,6 +25,7 @@ A minimal CLI tool to run Docker Compose commands across multiple hosts via SSH.
- [Installation](#installation)
- [Configuration](#configuration)
- [Multi-Host Services](#multi-host-services)
- [Config Command](#config-command)
- [Usage](#usage)
- [Auto-Migration](#auto-migration)
- [Traefik Multihost Ingress (File Provider)](#traefik-multihost-ingress-file-provider)
@@ -42,7 +43,7 @@ I used to run 100+ Docker Compose stacks on a single machine that kept running o
Both require changes to your compose files. **Compose Farm requires zero changes**—your existing `docker-compose.yml` files work as-is.
I also wanted a declarative setup—one config file that defines where everything runs. Change the config, run `up`, and services migrate automatically. See [Comparison with Alternatives](#comparison-with-alternatives) for how this compares to other approaches.
I also wanted a declarative setup—one config file that defines where everything runs. Change the config, run `cf apply`, and everything reconciles—services start, migrate, or stop as needed. See [Comparison with Alternatives](#comparison-with-alternatives) for how this compares to other approaches.
<p align="center">
<a href="https://xkcd.com/927/">
@@ -55,18 +56,26 @@ Before you say it—no, this is not a new standard. I changed nothing about my e
Compose Farm just automates what you'd do by hand:
- Runs `docker compose` commands over SSH
- Tracks which service runs on which host
- Auto-migrates services when you change the host assignment
- **One command (`cf apply`) to reconcile everything**—start missing services, migrate moved ones, stop removed ones
- Generates Traefik file-provider config for cross-host routing
**It's a convenience wrapper, not a new paradigm.**
## How It Works
1. You run `cf up plex`
2. Compose Farm looks up which host runs `plex` (e.g., `server-1`)
3. It SSHs to `server-1` (or runs locally if `localhost`)
4. It executes `docker compose -f /opt/compose/plex/docker-compose.yml up -d`
5. Output is streamed back with `[plex]` prefix
**The declarative way** — run `cf apply` and reality matches your config:
1. Compose Farm compares your config to what's actually running
2. Services in config but not running? **Starts them**
3. Services on the wrong host? **Migrates them** (stops on old host, starts on new)
4. Services running but removed from config? **Stops them**
**Under the hood** — each service operation is just SSH + docker compose:
1. Look up which host runs the service (e.g., `plex``server-1`)
2. SSH to `server-1` (or run locally if `localhost`)
3. Execute `docker compose -f /opt/compose/plex/docker-compose.yml up -d`
4. Stream output back with `[plex]` prefix
That's it. No orchestration, no service discovery, no magic.
@@ -134,6 +143,12 @@ pip install compose-farm
<details><summary>🐳 Docker</summary>
Using the provided `docker-compose.yml`:
```bash
docker compose run --rm cf up --all
```
Or directly:
```bash
docker run --rm \
-v $SSH_AUTH_SOCK:/ssh-agent -e SSH_AUTH_SOCK=/ssh-agent \
@@ -141,11 +156,6 @@ docker run --rm \
ghcr.io/basnijholt/compose-farm up --all
```
Or create an alias:
```bash
alias cf='docker run --rm -v $SSH_AUTH_SOCK:/ssh-agent -e SSH_AUTH_SOCK=/ssh-agent -v ./compose-farm.yaml:/root/.config/compose-farm/compose-farm.yaml:ro ghcr.io/basnijholt/compose-farm'
```
</details>
## Configuration
@@ -206,12 +216,27 @@ When you run `cf up autokuma`, it starts the service on all hosts in parallel. M
- Show output with `[service@host]` prefix for each host
- Track all running hosts in state
### Config Command
Compose Farm includes a `config` subcommand to help manage configuration files:
```bash
cf config init # Create a new config file with documented example
cf config show # Display current config with syntax highlighting
cf config path # Print the config file path (useful for scripting)
cf config validate # Validate config syntax and schema
cf config edit # Open config in $EDITOR
```
Use `cf config init` to get started with a fully documented template.
## Usage
The CLI is available as both `compose-farm` and the shorter `cf` alias.
| Command | Description |
|---------|-------------|
| **`cf apply`** | **Make reality match config (start + migrate + stop orphans)** |
| `cf up <svc>` | Start service (auto-migrates if host changed) |
| `cf down <svc>` | Stop service |
| `cf restart <svc>` | down + up |
@@ -219,21 +244,28 @@ The CLI is available as both `compose-farm` and the shorter `cf` alias.
| `cf pull <svc>` | Pull latest images |
| `cf logs -f <svc>` | Follow logs |
| `cf ps` | Show status of all services |
| `cf sync` | Discover running services + capture image digests |
| `cf refresh` | Update state from running services |
| `cf check` | Validate config, mounts, networks |
| `cf init-network` | Create Docker network on hosts |
| `cf traefik-file` | Generate Traefik file-provider config |
| `cf config <cmd>` | Manage config files (init, show, path, validate, edit) |
All commands support `--all` to operate on all services.
```bash
# Start services (auto-migrates if host changed in config)
cf up plex jellyfin
cf up --all
cf up --migrate # only services needing migration (state ≠ config)
Each command replaces: look up host → SSH → find compose file → run `ssh host "cd /opt/compose/plex && docker compose up -d"`.
# Stop services
cf down plex
```bash
# The main command: make reality match your config
cf apply # start missing + migrate + stop orphans
cf apply --dry-run # preview what would change
cf apply --no-orphans # skip stopping orphaned services
cf apply --full # also refresh all services (picks up config changes)
# Or operate on individual services
cf up plex jellyfin # start services (auto-migrates if host changed)
cf up --all
cf down plex # stop services
cf down --orphaned # stop services removed from config
# Pull latest images
cf pull --all
@@ -244,9 +276,9 @@ cf restart plex
# Update (pull + down + up) - the end-to-end update command
cf update --all
# Sync state with reality (discovers running services + captures image digests)
cf sync # updates state.yaml and dockerfarm-log.toml
cf sync --dry-run # preview without writing
# Update state from reality (discovers running services + captures digests)
cf refresh # updates state.yaml and dockerfarm-log.toml
cf refresh --dry-run # preview without writing
# Validate config, traefik labels, mounts, and networks
cf check # full validation (includes SSH checks)
@@ -265,6 +297,61 @@ cf logs -f plex # follow
cf ps
```
<details>
<summary>See the output of <code>cf --help</code></summary>
<!-- CODE:BASH:START -->
<!-- echo '```yaml' -->
<!-- export NO_COLOR=1 -->
<!-- export TERM=dumb -->
<!-- export TERMINAL_WIDTH=90 -->
<!-- cf --help -->
<!-- echo '```' -->
<!-- CODE:END -->
<!-- OUTPUT:START -->
<!-- ⚠️ This content is auto-generated by `markdown-code-runner`. -->
```yaml
Usage: cf [OPTIONS] COMMAND [ARGS]...
Compose Farm - run docker compose commands across multiple hosts
╭─ Options ────────────────────────────────────────────────────────────────────╮
│ --version -v Show version and exit │
│ --install-completion Install completion for the current shell. │
│ --show-completion Show completion for the current shell, to │
│ copy it or customize the installation. │
│ --help -h Show this message and exit. │
╰──────────────────────────────────────────────────────────────────────────────╯
╭─ Lifecycle ──────────────────────────────────────────────────────────────────╮
│ up Start services (docker compose up -d). Auto-migrates if host │
│ changed. │
│ down Stop services (docker compose down). │
│ pull Pull latest images (docker compose pull). │
│ restart Restart services (down + up). │
│ update Update services (pull + build + down + up). │
│ apply Make reality match config (start, migrate, stop as needed). │
╰──────────────────────────────────────────────────────────────────────────────╯
╭─ Configuration ──────────────────────────────────────────────────────────────╮
│ traefik-file Generate a Traefik file-provider fragment from compose │
│ Traefik labels. │
│ refresh Update local state from running services. │
│ check Validate configuration, traefik labels, mounts, and networks. │
│ init-network Create Docker network on hosts with consistent settings. │
│ config Manage compose-farm configuration files. │
╰──────────────────────────────────────────────────────────────────────────────╯
╭─ Monitoring ─────────────────────────────────────────────────────────────────╮
│ logs Show service logs. │
│ ps Show status of all services. │
│ stats Show overview statistics for hosts and services. │
╰──────────────────────────────────────────────────────────────────────────────╯
```
<!-- OUTPUT:END -->
</details>
### Auto-Migration
When you change a service's host assignment in config and run `up`, Compose Farm automatically:
@@ -273,7 +360,7 @@ When you change a service's host assignment in config and run `up`, Compose Farm
3. Runs `up -d` on the new host
4. Updates state tracking
Use `cf up --migrate` (or `-m`) to automatically find and migrate all services where the current state differs from config—no need to list them manually.
Use `cf apply` to automatically reconcile all services—it finds and migrates services on wrong hosts, stops orphaned services, and starts missing services.
```yaml
# Before: plex runs on server-1
@@ -285,6 +372,14 @@ services:
plex: server-2 # Compose Farm will migrate automatically
```
**Orphaned services**: When you remove (or comment out) a service from config, it becomes "orphaned"—tracked in state but no longer in config. Use these commands to handle orphans:
- `cf apply` — Migrate services AND stop orphans (the full reconcile)
- `cf down --orphaned` — Only stop orphaned services
- `cf apply --dry-run` — Preview what would change before applying
This makes the config truly declarative: comment out a service, run `cf apply`, and it stops.
## Traefik Multihost Ingress (File Provider)
If you run a single Traefik instance on one "frontdoor" host and want it to route to
@@ -389,16 +484,16 @@ Update your Traefik config to use directory watching instead of a single file:
There are many ways to run containers on multiple hosts. Here is where Compose Farm sits:
| | Docker Contexts | K8s / Swarm | Ansible / Terraform | Portainer / Coolify | Compose Farm |
| | Compose Farm | Docker Contexts | K8s / Swarm | Ansible / Terraform | Portainer / Coolify |
|---|:---:|:---:|:---:|:---:|:---:|
| No compose rewrites | ✅ | | | ✅ | ✅ |
| Version controlled | ✅ | ✅ | ✅ | | |
| State tracking | | | ✅ | ✅ | ✅ |
| Auto-migration | ❌ | ✅ | ❌ | ❌ | ✅ |
| Interactive CLI | | ❌ | ❌ | ❌ | |
| Parallel execution | | | ✅ | ✅ | ✅ |
| Agentless | ✅ | ❌ | ✅ | ❌ | ✅ |
| High availability | ❌ | | | ❌ | ❌ |
| No compose rewrites | ✅ | | | ✅ | ✅ |
| Version controlled | ✅ | ✅ | ✅ | | |
| State tracking | | | ✅ | ✅ | ✅ |
| Auto-migration | ✅ | ❌ | ✅ | ❌ | ❌ |
| Interactive CLI | | ❌ | ❌ | ❌ | |
| Parallel execution | | | ✅ | ✅ | ✅ |
| Agentless | ✅ | ✅ | ❌ | ✅ | ❌ |
| High availability | ❌ | | | ❌ | ❌ |
**Docker Contexts** — You can use `docker context create remote ssh://...` and `docker compose --context remote up`. But it's manual: you must remember which host runs which service, there's no global view, no parallel execution, and no auto-migration.

11
docker-compose.yml Normal file
View File

@@ -0,0 +1,11 @@
services:
cf:
image: ghcr.io/basnijholt/compose-farm:latest
volumes:
- ${SSH_AUTH_SOCK}:/ssh-agent:ro
# Compose directory (contains compose files AND compose-farm.yaml config)
- ${CF_COMPOSE_DIR:-/opt/compose}:${CF_COMPOSE_DIR:-/opt/compose}
environment:
- SSH_AUTH_SOCK=/ssh-agent
# Config file path (state stored alongside it)
- CF_CONFIG=${CF_COMPOSE_DIR:-/opt/compose}/compose-farm.yaml

View File

@@ -11,7 +11,7 @@
name: paperless-ngx
services:
redis:
image: redis:7
image: redis:8
container_name: paperless-redis
restart: unless-stopped
networks:

View File

@@ -4,6 +4,7 @@ dynamic = ["version"]
description = "Compose Farm - run docker compose commands across multiple hosts"
readme = "README.md"
license = "MIT"
license-files = ["LICENSE"]
authors = [
{ name = "Bas Nijholt", email = "bas@nijho.lt" }
]
@@ -34,6 +35,7 @@ classifiers = [
"Programming Language :: Python :: 3.11",
"Programming Language :: Python :: 3.12",
"Programming Language :: Python :: 3.13",
"Programming Language :: Python :: 3.14",
"Topic :: System :: Systems Administration",
"Topic :: Utilities",
"Typing :: Typed",

81
reddit-post.md Normal file
View File

@@ -0,0 +1,81 @@
# Title options
- Multi-host Docker Compose without Kubernetes or file changes
- I built a CLI to run Docker Compose across hosts. Zero changes to your files.
---
# I made a CLI to run Docker Compose across multiple hosts without Kubernetes or Swarm
I've been running 100+ Docker Compose stacks on a single machine, and it kept running out of memory. I needed to spread services across multiple hosts, but:
- **Kubernetes** felt like overkill. I don't need pods, ingress controllers, or 10x more YAML.
- **Docker Swarm** is basically in maintenance mode.
- Both require rewriting my compose files.
So I built **Compose Farm**, a simple CLI that runs `docker compose` commands over SSH. No agents, no cluster setup, no changes to your existing compose files.
## How it works
One YAML file maps services to hosts:
```yaml
compose_dir: /opt/stacks
hosts:
nuc: 192.168.1.10
hp: 192.168.1.11
services:
plex: nuc
jellyfin: hp
sonarr: nuc
radarr: nuc
```
Then just:
```bash
cf up plex # runs on nuc via SSH
cf up --all # starts everything on their assigned hosts
cf logs -f plex # streams logs
cf ps # shows status across all hosts
```
## Auto-migration
Change a service's host in the config and run `cf up`. It stops the service on the old host and starts it on the new one. No manual SSH needed.
```yaml
# Before
plex: nuc
# After (just change this)
plex: hp
```
```bash
cf up plex # migrates automatically
```
## Requirements
- SSH key auth to your hosts
- Same paths on all hosts (I use NFS from my NAS)
- That's it. No agents, no daemons.
## What it doesn't do
- No high availability (if a host goes down, services don't auto-migrate)
- No overlay networking (containers on different hosts can't talk via Docker DNS)
- No service discovery
- No health checks or automatic restarts
It's a convenience wrapper around `docker compose` + SSH. If you need failover or cross-host container networking, you probably do need Swarm or Kubernetes.
## Links
- GitHub: https://github.com/basnijholt/compose-farm
- Install: `uv tool install compose-farm` or `pip install compose-farm`
Built this in 4 days because I was mass-SSHing into machines like a caveman. Happy to answer questions or take feedback!

File diff suppressed because it is too large Load Diff

View File

@@ -0,0 +1,19 @@
"""CLI interface using Typer."""
from __future__ import annotations
# Import command modules to trigger registration via @app.command() decorators
from compose_farm.cli import (
config, # noqa: F401
lifecycle, # noqa: F401
management, # noqa: F401
monitoring, # noqa: F401
)
# Import the shared app instance
from compose_farm.cli.app import app
__all__ = ["app"]
if __name__ == "__main__":
app()

View File

@@ -0,0 +1,42 @@
"""Shared Typer app instance."""
from __future__ import annotations
from typing import Annotated
import typer
from compose_farm import __version__
__all__ = ["app"]
def _version_callback(value: bool) -> None:
"""Print version and exit."""
if value:
typer.echo(f"compose-farm {__version__}")
raise typer.Exit
app = typer.Typer(
name="compose-farm",
help="Compose Farm - run docker compose commands across multiple hosts",
no_args_is_help=True,
context_settings={"help_option_names": ["-h", "--help"]},
)
@app.callback()
def main(
version: Annotated[
bool,
typer.Option(
"--version",
"-v",
help="Show version and exit",
callback=_version_callback,
is_eager=True,
),
] = False,
) -> None:
"""Compose Farm - run docker compose commands across multiple hosts."""

View File

@@ -0,0 +1,226 @@
"""Shared CLI helpers, options, and utilities."""
from __future__ import annotations
import asyncio
import contextlib
from pathlib import Path
from typing import TYPE_CHECKING, Annotated, TypeVar
import typer
from rich.progress import (
BarColumn,
MofNCompleteColumn,
Progress,
SpinnerColumn,
TaskID,
TextColumn,
TimeElapsedColumn,
)
from compose_farm.console import console, err_console
if TYPE_CHECKING:
from collections.abc import Callable, Coroutine, Generator
from compose_farm.config import Config
from compose_farm.executor import CommandResult
_T = TypeVar("_T")
# --- Shared CLI Options ---
ServicesArg = Annotated[
list[str] | None,
typer.Argument(help="Services to operate on"),
]
AllOption = Annotated[
bool,
typer.Option("--all", "-a", help="Run on all services"),
]
ConfigOption = Annotated[
Path | None,
typer.Option("--config", "-c", help="Path to config file"),
]
LogPathOption = Annotated[
Path | None,
typer.Option("--log-path", "-l", help="Path to Dockerfarm TOML log"),
]
HostOption = Annotated[
str | None,
typer.Option("--host", "-H", help="Filter to services on this host"),
]
# --- Constants (internal) ---
_MISSING_PATH_PREVIEW_LIMIT = 2
_STATS_PREVIEW_LIMIT = 3 # Max number of pending migrations to show by name
@contextlib.contextmanager
def progress_bar(label: str, total: int) -> Generator[tuple[Progress, TaskID], None, None]:
"""Create a standardized progress bar with consistent styling.
Yields (progress, task_id). Use progress.update(task_id, advance=1, description=...)
to advance.
"""
with Progress(
SpinnerColumn(),
TextColumn(f"[bold blue]{label}[/]"),
BarColumn(),
MofNCompleteColumn(),
TextColumn(""),
TimeElapsedColumn(),
TextColumn(""),
TextColumn("[progress.description]{task.description}"),
console=console,
transient=True,
) as progress:
task_id = progress.add_task("", total=total)
yield progress, task_id
def load_config_or_exit(config_path: Path | None) -> Config:
"""Load config or exit with a friendly error message."""
# Lazy import: pydantic adds ~50ms to startup, only load when actually needed
from compose_farm.config import load_config # noqa: PLC0415
try:
return load_config(config_path)
except FileNotFoundError as e:
err_console.print(f"[red]✗[/] {e}")
raise typer.Exit(1) from e
def get_services(
services: list[str],
all_services: bool,
config_path: Path | None,
) -> tuple[list[str], Config]:
"""Resolve service list and load config."""
config = load_config_or_exit(config_path)
if all_services:
return list(config.services.keys()), config
if not services:
err_console.print("[red]✗[/] Specify services or use --all")
raise typer.Exit(1)
return list(services), config
def run_async(coro: Coroutine[None, None, _T]) -> _T:
"""Run async coroutine."""
try:
return asyncio.run(coro)
except KeyboardInterrupt:
console.print("\n[yellow]Interrupted[/]")
raise typer.Exit(130) from None # Standard exit code for SIGINT
def report_results(results: list[CommandResult]) -> None:
"""Report command results and exit with appropriate code."""
succeeded = [r for r in results if r.success]
failed = [r for r in results if not r.success]
# Always print summary when there are multiple results
if len(results) > 1:
console.print() # Blank line before summary
if failed:
for r in failed:
err_console.print(
f"[red]✗[/] [cyan]{r.service}[/] failed with exit code {r.exit_code}"
)
console.print()
console.print(
f"[green]✓[/] {len(succeeded)}/{len(results)} services succeeded, "
f"[red]✗[/] {len(failed)} failed"
)
else:
console.print(f"[green]✓[/] All {len(results)} services succeeded")
elif failed:
# Single service failed
r = failed[0]
err_console.print(f"[red]✗[/] [cyan]{r.service}[/] failed with exit code {r.exit_code}")
if failed:
raise typer.Exit(1)
def maybe_regenerate_traefik(
cfg: Config,
results: list[CommandResult] | None = None,
) -> None:
"""Regenerate traefik config if traefik_file is configured.
If results are provided, skips regeneration if all services failed.
"""
if cfg.traefik_file is None:
return
# Skip if all services failed
if results and not any(r.success for r in results):
return
# Lazy import: traefik/yaml adds startup time, only load when traefik_file is configured
from compose_farm.traefik import ( # noqa: PLC0415
generate_traefik_config,
render_traefik_config,
)
try:
dynamic, warnings = generate_traefik_config(cfg, list(cfg.services.keys()))
new_content = render_traefik_config(dynamic)
# Check if content changed
old_content = ""
if cfg.traefik_file.exists():
old_content = cfg.traefik_file.read_text()
if new_content != old_content:
cfg.traefik_file.parent.mkdir(parents=True, exist_ok=True)
cfg.traefik_file.write_text(new_content)
console.print() # Ensure we're on a new line after streaming output
console.print(f"[green]✓[/] Traefik config updated: {cfg.traefik_file}")
for warning in warnings:
err_console.print(f"[yellow]![/] {warning}")
except (FileNotFoundError, ValueError) as exc:
err_console.print(f"[yellow]![/] Failed to update traefik config: {exc}")
def validate_host_for_service(cfg: Config, service: str, host: str) -> None:
"""Validate that a host is valid for a service."""
if host not in cfg.hosts:
err_console.print(f"[red]✗[/] Host '{host}' not found in config")
raise typer.Exit(1)
allowed_hosts = cfg.get_hosts(service)
if host not in allowed_hosts:
err_console.print(
f"[red]✗[/] Service '{service}' is not configured for host '{host}' "
f"(configured: {', '.join(allowed_hosts)})"
)
raise typer.Exit(1)
def run_host_operation(
cfg: Config,
svc_list: list[str],
host: str,
command: str,
action_verb: str,
state_callback: Callable[[Config, str, str], None],
) -> None:
"""Run an operation on a specific host for multiple services."""
from compose_farm.executor import run_compose_on_host # noqa: PLC0415
results: list[CommandResult] = []
for service in svc_list:
validate_host_for_service(cfg, service, host)
console.print(f"[cyan]\\[{service}][/] {action_verb} on [magenta]{host}[/]...")
result = run_async(run_compose_on_host(cfg, service, host, command, raw=True))
print() # Newline after raw output
results.append(result)
if result.success:
state_callback(cfg, service, host)
maybe_regenerate_traefik(cfg, results)
report_results(results)

View File

@@ -0,0 +1,323 @@
"""Configuration management commands for compose-farm."""
from __future__ import annotations
import os
import platform
import shlex
import shutil
import subprocess
from importlib import resources
from pathlib import Path
from typing import Annotated
import typer
from compose_farm.cli.app import app
from compose_farm.console import console, err_console
from compose_farm.paths import config_search_paths, default_config_path
config_app = typer.Typer(
name="config",
help="Manage compose-farm configuration files.",
no_args_is_help=True,
)
# --- CLI Options (same pattern as cli.py) ---
_PathOption = Annotated[
Path | None,
typer.Option("--path", "-p", help="Path to config file. Uses auto-detection if not specified."),
]
_ForceOption = Annotated[
bool,
typer.Option("--force", "-f", help="Overwrite existing config without confirmation."),
]
_RawOption = Annotated[
bool,
typer.Option("--raw", "-r", help="Output raw file contents (for copy-paste)."),
]
def _get_editor() -> str:
"""Get the user's preferred editor.
Checks $EDITOR, then $VISUAL, then falls back to platform defaults.
"""
for env_var in ("EDITOR", "VISUAL"):
editor = os.environ.get(env_var)
if editor:
return editor
if platform.system() == "Windows":
return "notepad"
# Try common editors on Unix-like systems
for editor in ("nano", "vim", "vi"):
if shutil.which(editor):
return editor
return "vi"
def _generate_template() -> str:
"""Generate a config template with documented schema."""
try:
template_file = resources.files("compose_farm") / "example-config.yaml"
return template_file.read_text(encoding="utf-8")
except FileNotFoundError as e:
err_console.print("[red]Example config template is missing from the package.[/red]")
err_console.print("Reinstall compose-farm or report this issue.")
raise typer.Exit(1) from e
def _get_config_file(path: Path | None) -> Path | None:
"""Resolve config path, or auto-detect from standard locations."""
if path:
return path.expanduser().resolve()
# Check environment variable
if env_path := os.environ.get("CF_CONFIG"):
p = Path(env_path)
if p.exists():
return p.resolve()
# Check standard locations
for p in config_search_paths():
if p.exists():
return p.resolve()
return None
@config_app.command("init")
def config_init(
path: _PathOption = None,
force: _ForceOption = False,
) -> None:
"""Create a new config file with documented example.
The generated config file serves as a template showing all available
options with explanatory comments.
"""
target_path = (path.expanduser().resolve() if path else None) or default_config_path()
if target_path.exists() and not force:
console.print(
f"[bold yellow]Config file already exists at:[/bold yellow] [cyan]{target_path}[/cyan]",
)
if not typer.confirm("Overwrite existing config file?"):
console.print("[dim]Aborted.[/dim]")
raise typer.Exit(0)
# Create parent directories
target_path.parent.mkdir(parents=True, exist_ok=True)
# Generate and write template
template_content = _generate_template()
target_path.write_text(template_content, encoding="utf-8")
console.print(f"[green]✓[/] Config file created at: {target_path}")
console.print("\n[dim]Edit the file to customize your settings:[/dim]")
console.print(" [cyan]cf config edit[/cyan]")
@config_app.command("edit")
def config_edit(
path: _PathOption = None,
) -> None:
"""Open the config file in your default editor.
The editor is determined by: $EDITOR > $VISUAL > platform default.
"""
config_file = _get_config_file(path)
if config_file is None:
console.print("[yellow]No config file found.[/yellow]")
console.print("\nRun [bold cyan]cf config init[/bold cyan] to create one.")
console.print("\nSearched locations:")
for p in config_search_paths():
console.print(f" - {p}")
raise typer.Exit(1)
if not config_file.exists():
console.print("[yellow]Config file not found.[/yellow]")
console.print(f"\nProvided path does not exist: [cyan]{config_file}[/cyan]")
console.print("\nRun [bold cyan]cf config init[/bold cyan] to create one.")
raise typer.Exit(1)
editor = _get_editor()
console.print(f"[dim]Opening {config_file} with {editor}...[/dim]")
try:
editor_cmd = shlex.split(editor, posix=os.name != "nt")
except ValueError as e:
err_console.print("[red]Invalid editor command. Check $EDITOR/$VISUAL.[/red]")
raise typer.Exit(1) from e
if not editor_cmd:
err_console.print("[red]Editor command is empty.[/red]")
raise typer.Exit(1)
try:
subprocess.run([*editor_cmd, str(config_file)], check=True)
except FileNotFoundError:
err_console.print(f"[red]Editor '{editor_cmd[0]}' not found.[/red]")
err_console.print("Set $EDITOR environment variable to your preferred editor.")
raise typer.Exit(1) from None
except subprocess.CalledProcessError as e:
err_console.print(f"[red]Editor exited with error code {e.returncode}[/red]")
raise typer.Exit(e.returncode) from None
@config_app.command("show")
def config_show(
path: _PathOption = None,
raw: _RawOption = False,
) -> None:
"""Display the config file location and contents."""
config_file = _get_config_file(path)
if config_file is None:
console.print("[yellow]No config file found.[/yellow]")
console.print("\nSearched locations:")
for p in config_search_paths():
status = "[green]exists[/green]" if p.exists() else "[dim]not found[/dim]"
console.print(f" - {p} ({status})")
console.print("\nRun [bold cyan]cf config init[/bold cyan] to create one.")
raise typer.Exit(0)
if not config_file.exists():
console.print("[yellow]Config file not found.[/yellow]")
console.print(f"\nProvided path does not exist: [cyan]{config_file}[/cyan]")
console.print("\nRun [bold cyan]cf config init[/bold cyan] to create one.")
raise typer.Exit(1)
content = config_file.read_text(encoding="utf-8")
if raw:
print(content, end="")
return
from rich.syntax import Syntax # noqa: PLC0415
console.print(f"[bold green]Config file:[/bold green] [cyan]{config_file}[/cyan]")
console.print()
syntax = Syntax(content, "yaml", theme="monokai", line_numbers=True, word_wrap=True)
console.print(syntax)
console.print()
console.print("[dim]Tip: Use -r for copy-paste friendly output[/dim]")
@config_app.command("path")
def config_path(
path: _PathOption = None,
) -> None:
"""Print the config file path (useful for scripting)."""
config_file = _get_config_file(path)
if config_file is None:
console.print("[yellow]No config file found.[/yellow]")
console.print("\nSearched locations:")
for p in config_search_paths():
status = "[green]exists[/green]" if p.exists() else "[dim]not found[/dim]"
console.print(f" - {p} ({status})")
raise typer.Exit(1)
# Just print the path for easy piping
print(config_file)
@config_app.command("validate")
def config_validate(
path: _PathOption = None,
) -> None:
"""Validate the config file syntax and schema."""
config_file = _get_config_file(path)
if config_file is None:
err_console.print("[red]✗[/] No config file found")
raise typer.Exit(1)
# Lazy import: pydantic adds ~50ms to startup, only load when actually needed
from compose_farm.config import load_config # noqa: PLC0415
try:
cfg = load_config(config_file)
except FileNotFoundError as e:
err_console.print(f"[red]✗[/] {e}")
raise typer.Exit(1) from e
except Exception as e:
err_console.print(f"[red]✗[/] Invalid config: {e}")
raise typer.Exit(1) from e
console.print(f"[green]✓[/] Valid config: {config_file}")
console.print(f" Hosts: {len(cfg.hosts)}")
console.print(f" Services: {len(cfg.services)}")
@config_app.command("symlink")
def config_symlink(
target: Annotated[
Path | None,
typer.Argument(help="Config file to link to. Defaults to ./compose-farm.yaml"),
] = None,
force: _ForceOption = False,
) -> None:
"""Create a symlink from the default config location to a config file.
This makes a local config file discoverable globally without copying.
Always uses absolute paths to avoid broken symlinks.
Examples:
cf config symlink # Link to ./compose-farm.yaml
cf config symlink /opt/compose/config.yaml # Link to specific file
"""
# Default to compose-farm.yaml in current directory
target_path = (target or Path("compose-farm.yaml")).expanduser().resolve()
if not target_path.exists():
err_console.print(f"[red]✗[/] Target config file not found: {target_path}")
raise typer.Exit(1)
if not target_path.is_file():
err_console.print(f"[red]✗[/] Target is not a file: {target_path}")
raise typer.Exit(1)
symlink_path = default_config_path()
# Check if symlink location already exists
if symlink_path.exists() or symlink_path.is_symlink():
if symlink_path.is_symlink():
current_target = symlink_path.resolve() if symlink_path.exists() else None
if current_target == target_path:
console.print(f"[green]✓[/] Symlink already points to: {target_path}")
return
# Update existing symlink
if not force:
existing = symlink_path.readlink()
console.print(f"[yellow]Symlink exists:[/] {symlink_path} -> {existing}")
if not typer.confirm(f"Update to point to {target_path}?"):
console.print("[dim]Aborted.[/dim]")
raise typer.Exit(0)
symlink_path.unlink()
else:
# Regular file exists
err_console.print(f"[red]✗[/] A regular file exists at: {symlink_path}")
err_console.print(" Back it up or remove it first, then retry.")
raise typer.Exit(1)
# Create parent directories
symlink_path.parent.mkdir(parents=True, exist_ok=True)
# Create symlink with absolute path
symlink_path.symlink_to(target_path)
console.print("[green]✓[/] Created symlink:")
console.print(f" {symlink_path}")
console.print(f" -> {target_path}")
# Register config subcommand on the shared app
app.add_typer(config_app, name="config", rich_help_panel="Configuration")

View File

@@ -0,0 +1,299 @@
"""Lifecycle commands: up, down, pull, restart, update, apply."""
from __future__ import annotations
from typing import TYPE_CHECKING, Annotated
import typer
if TYPE_CHECKING:
from compose_farm.config import Config
from compose_farm.cli.app import app
from compose_farm.cli.common import (
AllOption,
ConfigOption,
HostOption,
ServicesArg,
get_services,
load_config_or_exit,
maybe_regenerate_traefik,
report_results,
run_async,
run_host_operation,
)
from compose_farm.console import console, err_console
from compose_farm.executor import run_on_services, run_sequential_on_services
from compose_farm.operations import stop_orphaned_services, up_services
from compose_farm.state import (
add_service_to_host,
get_orphaned_services,
get_service_host,
get_services_needing_migration,
get_services_not_in_state,
remove_service,
remove_service_from_host,
)
@app.command(rich_help_panel="Lifecycle")
def up(
services: ServicesArg = None,
all_services: AllOption = False,
host: HostOption = None,
config: ConfigOption = None,
) -> None:
"""Start services (docker compose up -d). Auto-migrates if host changed."""
svc_list, cfg = get_services(services or [], all_services, config)
# Per-host operation: run on specific host only
if host:
run_host_operation(cfg, svc_list, host, "up -d", "Starting", add_service_to_host)
return
# Normal operation: use up_services with migration logic
results = run_async(up_services(cfg, svc_list, raw=True))
maybe_regenerate_traefik(cfg, results)
report_results(results)
@app.command(rich_help_panel="Lifecycle")
def down(
services: ServicesArg = None,
all_services: AllOption = False,
orphaned: Annotated[
bool,
typer.Option(
"--orphaned", help="Stop orphaned services (in state but removed from config)"
),
] = False,
host: HostOption = None,
config: ConfigOption = None,
) -> None:
"""Stop services (docker compose down)."""
# Handle --orphaned flag
if orphaned:
if services or all_services or host:
err_console.print("[red]✗[/] Cannot use --orphaned with services, --all, or --host")
raise typer.Exit(1)
cfg = load_config_or_exit(config)
orphaned_services = get_orphaned_services(cfg)
if not orphaned_services:
console.print("[green]✓[/] No orphaned services to stop")
return
console.print(
f"[yellow]Stopping {len(orphaned_services)} orphaned service(s):[/] "
f"{', '.join(orphaned_services.keys())}"
)
results = run_async(stop_orphaned_services(cfg))
report_results(results)
return
svc_list, cfg = get_services(services or [], all_services, config)
# Per-host operation: run on specific host only
if host:
run_host_operation(cfg, svc_list, host, "down", "Stopping", remove_service_from_host)
return
# Normal operation
raw = len(svc_list) == 1
results = run_async(run_on_services(cfg, svc_list, "down", raw=raw))
# Remove from state on success
# For multi-host services, result.service is "svc@host", extract base name
removed_services: set[str] = set()
for result in results:
if result.success:
base_service = result.service.split("@")[0]
if base_service not in removed_services:
remove_service(cfg, base_service)
removed_services.add(base_service)
maybe_regenerate_traefik(cfg, results)
report_results(results)
@app.command(rich_help_panel="Lifecycle")
def pull(
services: ServicesArg = None,
all_services: AllOption = False,
config: ConfigOption = None,
) -> None:
"""Pull latest images (docker compose pull)."""
svc_list, cfg = get_services(services or [], all_services, config)
raw = len(svc_list) == 1
results = run_async(run_on_services(cfg, svc_list, "pull", raw=raw))
report_results(results)
@app.command(rich_help_panel="Lifecycle")
def restart(
services: ServicesArg = None,
all_services: AllOption = False,
config: ConfigOption = None,
) -> None:
"""Restart services (down + up)."""
svc_list, cfg = get_services(services or [], all_services, config)
raw = len(svc_list) == 1
results = run_async(run_sequential_on_services(cfg, svc_list, ["down", "up -d"], raw=raw))
maybe_regenerate_traefik(cfg, results)
report_results(results)
@app.command(rich_help_panel="Lifecycle")
def update(
services: ServicesArg = None,
all_services: AllOption = False,
config: ConfigOption = None,
) -> None:
"""Update services (pull + build + down + up)."""
svc_list, cfg = get_services(services or [], all_services, config)
raw = len(svc_list) == 1
results = run_async(
run_sequential_on_services(
cfg, svc_list, ["pull --ignore-buildable", "build", "down", "up -d"], raw=raw
)
)
maybe_regenerate_traefik(cfg, results)
report_results(results)
def _format_host(host: str | list[str]) -> str:
"""Format a host value for display."""
if isinstance(host, list):
return ", ".join(host)
return host
def _report_pending_migrations(cfg: Config, migrations: list[str]) -> None:
"""Report services that need migration."""
console.print(f"[cyan]Services to migrate ({len(migrations)}):[/]")
for svc in migrations:
current = get_service_host(cfg, svc)
target = cfg.get_hosts(svc)[0]
console.print(f" [cyan]{svc}[/]: [magenta]{current}[/] → [magenta]{target}[/]")
def _report_pending_orphans(orphaned: dict[str, str | list[str]]) -> None:
"""Report orphaned services that will be stopped."""
console.print(f"[yellow]Orphaned services to stop ({len(orphaned)}):[/]")
for svc, hosts in orphaned.items():
console.print(f" [cyan]{svc}[/] on [magenta]{_format_host(hosts)}[/]")
def _report_pending_starts(cfg: Config, missing: list[str]) -> None:
"""Report services that will be started."""
console.print(f"[green]Services to start ({len(missing)}):[/]")
for svc in missing:
target = _format_host(cfg.get_hosts(svc))
console.print(f" [cyan]{svc}[/] on [magenta]{target}[/]")
def _report_pending_refresh(cfg: Config, to_refresh: list[str]) -> None:
"""Report services that will be refreshed."""
console.print(f"[blue]Services to refresh ({len(to_refresh)}):[/]")
for svc in to_refresh:
target = _format_host(cfg.get_hosts(svc))
console.print(f" [cyan]{svc}[/] on [magenta]{target}[/]")
@app.command(rich_help_panel="Lifecycle")
def apply(
dry_run: Annotated[
bool,
typer.Option("--dry-run", "-n", help="Show what would change without executing"),
] = False,
no_orphans: Annotated[
bool,
typer.Option("--no-orphans", help="Only migrate, don't stop orphaned services"),
] = False,
full: Annotated[
bool,
typer.Option("--full", "-f", help="Also run up on all services to apply config changes"),
] = False,
config: ConfigOption = None,
) -> None:
"""Make reality match config (start, migrate, stop as needed).
This is the "reconcile" command that ensures running services match your
config file. It will:
1. Stop orphaned services (in state but removed from config)
2. Migrate services on wrong host (host in state ≠ host in config)
3. Start missing services (in config but not in state)
Use --dry-run to preview changes before applying.
Use --no-orphans to only migrate/start without stopping orphaned services.
Use --full to also run 'up' on all services (picks up compose/env changes).
"""
cfg = load_config_or_exit(config)
orphaned = get_orphaned_services(cfg)
migrations = get_services_needing_migration(cfg)
missing = get_services_not_in_state(cfg)
# For --full: refresh all services not already being started/migrated
handled = set(migrations) | set(missing)
to_refresh = [svc for svc in cfg.services if svc not in handled] if full else []
has_orphans = bool(orphaned) and not no_orphans
has_migrations = bool(migrations)
has_missing = bool(missing)
has_refresh = bool(to_refresh)
if not has_orphans and not has_migrations and not has_missing and not has_refresh:
console.print("[green]✓[/] Nothing to apply - reality matches config")
return
# Report what will be done
if has_orphans:
_report_pending_orphans(orphaned)
if has_migrations:
_report_pending_migrations(cfg, migrations)
if has_missing:
_report_pending_starts(cfg, missing)
if has_refresh:
_report_pending_refresh(cfg, to_refresh)
if dry_run:
console.print("\n[dim](dry-run: no changes made)[/]")
return
# Execute changes
console.print()
all_results = []
# 1. Stop orphaned services first
if has_orphans:
console.print("[yellow]Stopping orphaned services...[/]")
all_results.extend(run_async(stop_orphaned_services(cfg)))
# 2. Migrate services on wrong host
if has_migrations:
console.print("[cyan]Migrating services...[/]")
migrate_results = run_async(up_services(cfg, migrations, raw=True))
all_results.extend(migrate_results)
maybe_regenerate_traefik(cfg, migrate_results)
# 3. Start missing services (reuse up_services which handles state updates)
if has_missing:
console.print("[green]Starting missing services...[/]")
start_results = run_async(up_services(cfg, missing, raw=True))
all_results.extend(start_results)
maybe_regenerate_traefik(cfg, start_results)
# 4. Refresh remaining services (--full: run up to apply config changes)
if has_refresh:
console.print("[blue]Refreshing services...[/]")
refresh_results = run_async(up_services(cfg, to_refresh, raw=True))
all_results.extend(refresh_results)
maybe_regenerate_traefik(cfg, refresh_results)
report_results(all_results)
# Alias: cf a = cf apply
app.command("a", hidden=True)(apply)

View File

@@ -0,0 +1,658 @@
"""Management commands: sync, check, init-network, traefik-file."""
from __future__ import annotations
import asyncio
from datetime import UTC, datetime
from pathlib import Path # noqa: TC003
from typing import TYPE_CHECKING, Annotated
import typer
from rich.progress import Progress, TaskID # noqa: TC002
from compose_farm.cli.app import app
from compose_farm.cli.common import (
_MISSING_PATH_PREVIEW_LIMIT,
AllOption,
ConfigOption,
LogPathOption,
ServicesArg,
get_services,
load_config_or_exit,
progress_bar,
run_async,
)
if TYPE_CHECKING:
from compose_farm.config import Config
from compose_farm.console import console, err_console
from compose_farm.executor import (
CommandResult,
check_service_running,
is_local,
run_command,
)
from compose_farm.logs import (
DEFAULT_LOG_PATH,
SnapshotEntry,
collect_service_entries,
isoformat,
load_existing_entries,
merge_entries,
write_toml,
)
from compose_farm.operations import check_host_compatibility, check_service_requirements
from compose_farm.state import get_orphaned_services, load_state, save_state
from compose_farm.traefik import generate_traefik_config, render_traefik_config
# --- Sync helpers ---
def _discover_services(cfg: Config) -> dict[str, str | list[str]]:
"""Discover running services with a progress bar."""
async def check_service(service: str) -> tuple[str, str | list[str] | None]:
"""Check where a service is running.
For multi-host services, returns list of hosts where running.
For single-host, returns single host name or None.
"""
assigned_hosts = cfg.get_hosts(service)
if cfg.is_multi_host(service):
# Multi-host: find all hosts where running (check in parallel)
checks = await asyncio.gather(
*[check_service_running(cfg, service, h) for h in assigned_hosts]
)
running_hosts = [
h for h, running in zip(assigned_hosts, checks, strict=True) if running
]
return service, running_hosts if running_hosts else None
# Single-host: check assigned host first
assigned_host = assigned_hosts[0]
if await check_service_running(cfg, service, assigned_host):
return service, assigned_host
# Check other hosts
for host_name in cfg.hosts:
if host_name == assigned_host:
continue
if await check_service_running(cfg, service, host_name):
return service, host_name
return service, None
async def gather_with_progress(
progress: Progress, task_id: TaskID
) -> dict[str, str | list[str]]:
services = list(cfg.services.keys())
tasks = [asyncio.create_task(check_service(s)) for s in services]
discovered: dict[str, str | list[str]] = {}
for coro in asyncio.as_completed(tasks):
service, host = await coro
if host is not None:
discovered[service] = host
progress.update(task_id, advance=1, description=f"[cyan]{service}[/]")
return discovered
with progress_bar("Discovering", len(cfg.services)) as (progress, task_id):
return asyncio.run(gather_with_progress(progress, task_id))
def _snapshot_services(
cfg: Config,
services: list[str],
log_path: Path | None,
) -> Path:
"""Capture image digests with a progress bar."""
async def collect_service(service: str, now: datetime) -> list[SnapshotEntry]:
try:
return await collect_service_entries(cfg, service, now=now)
except RuntimeError:
return []
async def gather_with_progress(
progress: Progress, task_id: TaskID, now: datetime, svc_list: list[str]
) -> list[SnapshotEntry]:
# Map tasks to service names so we can update description
task_to_service = {asyncio.create_task(collect_service(s, now)): s for s in svc_list}
all_entries: list[SnapshotEntry] = []
for coro in asyncio.as_completed(list(task_to_service.keys())):
entries = await coro
all_entries.extend(entries)
# Find which service just completed (by checking done tasks)
for t, svc in task_to_service.items():
if t.done() and not hasattr(t, "_reported"):
t._reported = True # type: ignore[attr-defined]
progress.update(task_id, advance=1, description=f"[cyan]{svc}[/]")
break
return all_entries
effective_log_path = log_path or DEFAULT_LOG_PATH
now_dt = datetime.now(UTC)
now_iso = isoformat(now_dt)
with progress_bar("Capturing", len(services)) as (progress, task_id):
snapshot_entries = asyncio.run(gather_with_progress(progress, task_id, now_dt, services))
if not snapshot_entries:
msg = "No image digests were captured"
raise RuntimeError(msg)
existing_entries = load_existing_entries(effective_log_path)
merged_entries = merge_entries(existing_entries, snapshot_entries, now_iso=now_iso)
meta = {"generated_at": now_iso, "compose_dir": str(cfg.compose_dir)}
write_toml(effective_log_path, meta=meta, entries=merged_entries)
return effective_log_path
def _format_host(host: str | list[str]) -> str:
"""Format a host value for display."""
if isinstance(host, list):
return ", ".join(host)
return host
def _report_sync_changes(
added: list[str],
removed: list[str],
changed: list[tuple[str, str | list[str], str | list[str]]],
discovered: dict[str, str | list[str]],
current_state: dict[str, str | list[str]],
) -> None:
"""Report sync changes to the user."""
if added:
console.print(f"\nNew services found ({len(added)}):")
for service in sorted(added):
host_str = _format_host(discovered[service])
console.print(f" [green]+[/] [cyan]{service}[/] on [magenta]{host_str}[/]")
if changed:
console.print(f"\nServices on different hosts ({len(changed)}):")
for service, old_host, new_host in sorted(changed):
old_str = _format_host(old_host)
new_str = _format_host(new_host)
console.print(
f" [yellow]~[/] [cyan]{service}[/]: [magenta]{old_str}[/] → [magenta]{new_str}[/]"
)
if removed:
console.print(f"\nServices no longer running ({len(removed)}):")
for service in sorted(removed):
host_str = _format_host(current_state[service])
console.print(f" [red]-[/] [cyan]{service}[/] (was on [magenta]{host_str}[/])")
# --- Check helpers ---
def _check_ssh_connectivity(cfg: Config) -> list[str]:
"""Check SSH connectivity to all hosts. Returns list of unreachable hosts."""
# Filter out local hosts - no SSH needed
remote_hosts = [h for h in cfg.hosts if not is_local(cfg.hosts[h])]
if not remote_hosts:
return []
console.print() # Spacing before progress bar
async def check_host(host_name: str) -> tuple[str, bool]:
host = cfg.hosts[host_name]
result = await run_command(host, "echo ok", host_name, stream=False)
return host_name, result.success
async def gather_with_progress(progress: Progress, task_id: TaskID) -> list[str]:
tasks = [asyncio.create_task(check_host(h)) for h in remote_hosts]
unreachable: list[str] = []
for coro in asyncio.as_completed(tasks):
host_name, success = await coro
if not success:
unreachable.append(host_name)
progress.update(task_id, advance=1, description=f"[cyan]{host_name}[/]")
return unreachable
with progress_bar("Checking SSH connectivity", len(remote_hosts)) as (progress, task_id):
return asyncio.run(gather_with_progress(progress, task_id))
def _check_service_requirements(
cfg: Config,
services: list[str],
) -> tuple[list[tuple[str, str, str]], list[tuple[str, str, str]], list[tuple[str, str, str]]]:
"""Check mounts, networks, and devices for all services with a progress bar.
Returns (mount_errors, network_errors, device_errors) where each is a list of
(service, host, missing_item) tuples.
"""
async def check_service(
service: str,
) -> tuple[
str,
list[tuple[str, str, str]],
list[tuple[str, str, str]],
list[tuple[str, str, str]],
]:
"""Check requirements for a single service on all its hosts."""
host_names = cfg.get_hosts(service)
mount_errors: list[tuple[str, str, str]] = []
network_errors: list[tuple[str, str, str]] = []
device_errors: list[tuple[str, str, str]] = []
for host_name in host_names:
missing_paths, missing_nets, missing_devs = await check_service_requirements(
cfg, service, host_name
)
mount_errors.extend((service, host_name, p) for p in missing_paths)
network_errors.extend((service, host_name, n) for n in missing_nets)
device_errors.extend((service, host_name, d) for d in missing_devs)
return service, mount_errors, network_errors, device_errors
async def gather_with_progress(
progress: Progress, task_id: TaskID
) -> tuple[list[tuple[str, str, str]], list[tuple[str, str, str]], list[tuple[str, str, str]]]:
tasks = [asyncio.create_task(check_service(s)) for s in services]
all_mount_errors: list[tuple[str, str, str]] = []
all_network_errors: list[tuple[str, str, str]] = []
all_device_errors: list[tuple[str, str, str]] = []
for coro in asyncio.as_completed(tasks):
service, mount_errs, net_errs, dev_errs = await coro
all_mount_errors.extend(mount_errs)
all_network_errors.extend(net_errs)
all_device_errors.extend(dev_errs)
progress.update(task_id, advance=1, description=f"[cyan]{service}[/]")
return all_mount_errors, all_network_errors, all_device_errors
with progress_bar("Checking requirements", len(services)) as (progress, task_id):
return asyncio.run(gather_with_progress(progress, task_id))
def _report_config_status(cfg: Config) -> bool:
"""Check and report config vs disk status. Returns True if errors found."""
configured = set(cfg.services.keys())
on_disk = cfg.discover_compose_dirs()
unmanaged = sorted(on_disk - configured)
missing_from_disk = sorted(configured - on_disk)
if unmanaged:
console.print(f"\n[yellow]Unmanaged[/] (on disk but not in config, {len(unmanaged)}):")
for name in unmanaged:
console.print(f" [yellow]+[/] [cyan]{name}[/]")
if missing_from_disk:
console.print(f"\n[red]In config but no compose file[/] ({len(missing_from_disk)}):")
for name in missing_from_disk:
console.print(f" [red]-[/] [cyan]{name}[/]")
if not unmanaged and not missing_from_disk:
console.print("[green]✓[/] Config matches disk")
return bool(missing_from_disk)
def _report_orphaned_services(cfg: Config) -> bool:
"""Check for services in state but not in config. Returns True if orphans found."""
orphaned = get_orphaned_services(cfg)
if orphaned:
console.print("\n[yellow]Orphaned services[/] (in state but not in config):")
console.print(
"[dim]Run 'cf apply' to stop them, or 'cf down --orphaned' for just orphans.[/]"
)
for name, hosts in sorted(orphaned.items()):
host_str = ", ".join(hosts) if isinstance(hosts, list) else hosts
console.print(f" [yellow]![/] [cyan]{name}[/] on [magenta]{host_str}[/]")
return True
return False
def _report_traefik_status(cfg: Config, services: list[str]) -> None:
"""Check and report traefik label status."""
try:
_, warnings = generate_traefik_config(cfg, services, check_all=True)
except (FileNotFoundError, ValueError):
return
if warnings:
console.print(f"\n[yellow]Traefik issues[/] ({len(warnings)}):")
for warning in warnings:
console.print(f" [yellow]![/] {warning}")
else:
console.print("[green]✓[/] Traefik labels valid")
def _report_mount_errors(mount_errors: list[tuple[str, str, str]]) -> None:
"""Report mount errors grouped by service."""
by_service: dict[str, list[tuple[str, str]]] = {}
for svc, host, path in mount_errors:
by_service.setdefault(svc, []).append((host, path))
console.print(f"[red]Missing mounts[/] ({len(mount_errors)}):")
for svc, items in sorted(by_service.items()):
host = items[0][0]
paths = [p for _, p in items]
console.print(f" [cyan]{svc}[/] on [magenta]{host}[/]:")
for path in paths:
console.print(f" [red]✗[/] {path}")
def _report_network_errors(network_errors: list[tuple[str, str, str]]) -> None:
"""Report network errors grouped by service."""
by_service: dict[str, list[tuple[str, str]]] = {}
for svc, host, net in network_errors:
by_service.setdefault(svc, []).append((host, net))
console.print(f"[red]Missing networks[/] ({len(network_errors)}):")
for svc, items in sorted(by_service.items()):
host = items[0][0]
networks = [n for _, n in items]
console.print(f" [cyan]{svc}[/] on [magenta]{host}[/]:")
for net in networks:
console.print(f" [red]✗[/] {net}")
def _report_device_errors(device_errors: list[tuple[str, str, str]]) -> None:
"""Report device errors grouped by service."""
by_service: dict[str, list[tuple[str, str]]] = {}
for svc, host, dev in device_errors:
by_service.setdefault(svc, []).append((host, dev))
console.print(f"[red]Missing devices[/] ({len(device_errors)}):")
for svc, items in sorted(by_service.items()):
host = items[0][0]
devices = [d for _, d in items]
console.print(f" [cyan]{svc}[/] on [magenta]{host}[/]:")
for dev in devices:
console.print(f" [red]✗[/] {dev}")
def _report_ssh_status(unreachable_hosts: list[str]) -> bool:
"""Report SSH connectivity status. Returns True if there are errors."""
if unreachable_hosts:
console.print(f"[red]Unreachable hosts[/] ({len(unreachable_hosts)}):")
for host in sorted(unreachable_hosts):
console.print(f" [red]✗[/] [magenta]{host}[/]")
return True
console.print("[green]✓[/] All hosts reachable")
return False
def _report_host_compatibility(
compat: dict[str, tuple[int, int, list[str]]],
assigned_hosts: list[str],
) -> None:
"""Report host compatibility for a service."""
for host_name, (found, total, missing) in sorted(compat.items()):
is_assigned = host_name in assigned_hosts
marker = " [dim](assigned)[/]" if is_assigned else ""
if found == total:
console.print(f" [green]✓[/] [magenta]{host_name}[/] {found}/{total}{marker}")
else:
preview = ", ".join(missing[:_MISSING_PATH_PREVIEW_LIMIT])
if len(missing) > _MISSING_PATH_PREVIEW_LIMIT:
preview += f", +{len(missing) - _MISSING_PATH_PREVIEW_LIMIT} more"
console.print(
f" [red]✗[/] [magenta]{host_name}[/] {found}/{total} "
f"[dim](missing: {preview})[/]{marker}"
)
def _run_remote_checks(cfg: Config, svc_list: list[str], *, show_host_compat: bool) -> bool:
"""Run SSH-based checks for mounts, networks, and host compatibility.
Returns True if any errors were found.
"""
has_errors = False
# Check SSH connectivity first
if _report_ssh_status(_check_ssh_connectivity(cfg)):
has_errors = True
console.print() # Spacing before mounts/networks check
# Check mounts, networks, and devices
mount_errors, network_errors, device_errors = _check_service_requirements(cfg, svc_list)
if mount_errors:
_report_mount_errors(mount_errors)
has_errors = True
if network_errors:
_report_network_errors(network_errors)
has_errors = True
if device_errors:
_report_device_errors(device_errors)
has_errors = True
if not mount_errors and not network_errors and not device_errors:
console.print("[green]✓[/] All mounts, networks, and devices exist")
if show_host_compat:
for service in svc_list:
console.print(f"\n[bold]Host compatibility for[/] [cyan]{service}[/]:")
compat = run_async(check_host_compatibility(cfg, service))
assigned_hosts = cfg.get_hosts(service)
_report_host_compatibility(compat, assigned_hosts)
return has_errors
# Default network settings for cross-host Docker networking
_DEFAULT_NETWORK_NAME = "mynetwork"
_DEFAULT_NETWORK_SUBNET = "172.20.0.0/16"
_DEFAULT_NETWORK_GATEWAY = "172.20.0.1"
@app.command("traefik-file", rich_help_panel="Configuration")
def traefik_file(
services: ServicesArg = None,
all_services: AllOption = False,
output: Annotated[
Path | None,
typer.Option(
"--output",
"-o",
help="Write Traefik file-provider YAML to this path (stdout if omitted)",
),
] = None,
config: ConfigOption = None,
) -> None:
"""Generate a Traefik file-provider fragment from compose Traefik labels."""
svc_list, cfg = get_services(services or [], all_services, config)
try:
dynamic, warnings = generate_traefik_config(cfg, svc_list)
except (FileNotFoundError, ValueError) as exc:
err_console.print(f"[red]✗[/] {exc}")
raise typer.Exit(1) from exc
rendered = render_traefik_config(dynamic)
if output:
output.parent.mkdir(parents=True, exist_ok=True)
output.write_text(rendered)
console.print(f"[green]✓[/] Traefik config written to {output}")
else:
console.print(rendered)
for warning in warnings:
err_console.print(f"[yellow]![/] {warning}")
@app.command(rich_help_panel="Configuration")
def refresh(
config: ConfigOption = None,
log_path: LogPathOption = None,
dry_run: Annotated[
bool,
typer.Option("--dry-run", "-n", help="Show what would change without writing"),
] = False,
) -> None:
"""Update local state from running services.
Discovers which services are running on which hosts, updates the state
file, and captures image digests. This is a read operation - it updates
your local state to match reality, not the other way around.
Use 'cf apply' to make reality match your config (stop orphans, migrate).
"""
cfg = load_config_or_exit(config)
current_state = load_state(cfg)
discovered = _discover_services(cfg)
# Calculate changes
added = [s for s in discovered if s not in current_state]
removed = [s for s in current_state if s not in discovered]
changed = [
(s, current_state[s], discovered[s])
for s in discovered
if s in current_state and current_state[s] != discovered[s]
]
# Report state changes
state_changed = bool(added or removed or changed)
if state_changed:
_report_sync_changes(added, removed, changed, discovered, current_state)
else:
console.print("[green]✓[/] State is already in sync.")
if dry_run:
console.print("\n[dim](dry-run: no changes made)[/]")
return
# Update state file
if state_changed:
save_state(cfg, discovered)
console.print(f"\n[green]✓[/] State updated: {len(discovered)} services tracked.")
# Capture image digests for running services
if discovered:
try:
path = _snapshot_services(cfg, list(discovered.keys()), log_path)
console.print(f"[green]✓[/] Digests written to {path}")
except RuntimeError as exc:
err_console.print(f"[yellow]![/] {exc}")
@app.command(rich_help_panel="Configuration")
def check(
services: ServicesArg = None,
local: Annotated[
bool,
typer.Option("--local", help="Skip SSH-based checks (faster)"),
] = False,
config: ConfigOption = None,
) -> None:
"""Validate configuration, traefik labels, mounts, and networks.
Without arguments: validates all services against configured hosts.
With service arguments: validates specific services and shows host compatibility.
Use --local to skip SSH-based checks for faster validation.
"""
cfg = load_config_or_exit(config)
# Determine which services to check and whether to show host compatibility
if services:
svc_list = list(services)
invalid = [s for s in svc_list if s not in cfg.services]
if invalid:
for svc in invalid:
err_console.print(f"[red]✗[/] Service '{svc}' not found in config")
raise typer.Exit(1)
show_host_compat = True
else:
svc_list = list(cfg.services.keys())
show_host_compat = False
# Run checks
has_errors = _report_config_status(cfg)
_report_traefik_status(cfg, svc_list)
if not local and _run_remote_checks(cfg, svc_list, show_host_compat=show_host_compat):
has_errors = True
# Check for orphaned services (in state but removed from config)
if _report_orphaned_services(cfg):
has_errors = True
if has_errors:
raise typer.Exit(1)
@app.command("init-network", rich_help_panel="Configuration")
def init_network(
hosts: Annotated[
list[str] | None,
typer.Argument(help="Hosts to create network on (default: all)"),
] = None,
network: Annotated[
str,
typer.Option("--network", "-n", help="Network name"),
] = _DEFAULT_NETWORK_NAME,
subnet: Annotated[
str,
typer.Option("--subnet", "-s", help="Network subnet"),
] = _DEFAULT_NETWORK_SUBNET,
gateway: Annotated[
str,
typer.Option("--gateway", "-g", help="Network gateway"),
] = _DEFAULT_NETWORK_GATEWAY,
config: ConfigOption = None,
) -> None:
"""Create Docker network on hosts with consistent settings.
Creates an external Docker network that services can use for cross-host
communication. Uses the same subnet/gateway on all hosts to ensure
consistent networking.
"""
cfg = load_config_or_exit(config)
target_hosts = list(hosts) if hosts else list(cfg.hosts.keys())
invalid = [h for h in target_hosts if h not in cfg.hosts]
if invalid:
for h in invalid:
err_console.print(f"[red]✗[/] Host '{h}' not found in config")
raise typer.Exit(1)
async def create_network_on_host(host_name: str) -> CommandResult:
host = cfg.hosts[host_name]
# Check if network already exists
check_cmd = f"docker network inspect '{network}' >/dev/null 2>&1"
check_result = await run_command(host, check_cmd, host_name, stream=False)
if check_result.success:
console.print(f"[cyan]\\[{host_name}][/] Network '{network}' already exists")
return CommandResult(service=host_name, exit_code=0, success=True)
# Create the network
create_cmd = (
f"docker network create "
f"--driver bridge "
f"--subnet '{subnet}' "
f"--gateway '{gateway}' "
f"'{network}'"
)
result = await run_command(host, create_cmd, host_name, stream=False)
if result.success:
console.print(f"[cyan]\\[{host_name}][/] [green]✓[/] Created network '{network}'")
else:
err_console.print(
f"[cyan]\\[{host_name}][/] [red]✗[/] Failed to create network: "
f"{result.stderr.strip()}"
)
return result
async def run_all() -> list[CommandResult]:
return await asyncio.gather(*[create_network_on_host(h) for h in target_hosts])
results = run_async(run_all())
failed = [r for r in results if not r.success]
if failed:
raise typer.Exit(1)

View File

@@ -0,0 +1,236 @@
"""Monitoring commands: logs, ps, stats."""
from __future__ import annotations
import asyncio
import contextlib
from typing import TYPE_CHECKING, Annotated
import typer
from rich.progress import Progress, TaskID # noqa: TC002
from rich.table import Table
from compose_farm.cli.app import app
from compose_farm.cli.common import (
_STATS_PREVIEW_LIMIT,
AllOption,
ConfigOption,
HostOption,
ServicesArg,
get_services,
load_config_or_exit,
progress_bar,
report_results,
run_async,
)
from compose_farm.console import console, err_console
from compose_farm.executor import run_command, run_on_services
from compose_farm.state import get_services_needing_migration, load_state
if TYPE_CHECKING:
from collections.abc import Mapping
from compose_farm.config import Config
def _group_services_by_host(
services: dict[str, str | list[str]],
hosts: Mapping[str, object],
all_hosts: list[str] | None = None,
) -> dict[str, list[str]]:
"""Group services by their assigned host(s).
For multi-host services (list or "all"), the service appears in multiple host lists.
"""
by_host: dict[str, list[str]] = {h: [] for h in hosts}
for service, host_value in services.items():
if isinstance(host_value, list):
# Explicit list of hosts
for host_name in host_value:
if host_name in by_host:
by_host[host_name].append(service)
elif host_value == "all" and all_hosts:
# "all" keyword - add to all hosts
for host_name in all_hosts:
if host_name in by_host:
by_host[host_name].append(service)
elif host_value in by_host:
# Single host
by_host[host_value].append(service)
return by_host
def _get_container_counts(cfg: Config) -> dict[str, int]:
"""Get container counts from all hosts with a progress bar."""
async def get_count(host_name: str) -> tuple[str, int]:
host = cfg.hosts[host_name]
result = await run_command(host, "docker ps -q | wc -l", host_name, stream=False)
count = 0
if result.success:
with contextlib.suppress(ValueError):
count = int(result.stdout.strip())
return host_name, count
async def gather_with_progress(progress: Progress, task_id: TaskID) -> dict[str, int]:
hosts = list(cfg.hosts.keys())
tasks = [asyncio.create_task(get_count(h)) for h in hosts]
results: dict[str, int] = {}
for coro in asyncio.as_completed(tasks):
host_name, count = await coro
results[host_name] = count
progress.update(task_id, advance=1, description=f"[cyan]{host_name}[/]")
return results
with progress_bar("Querying hosts", len(cfg.hosts)) as (progress, task_id):
return asyncio.run(gather_with_progress(progress, task_id))
def _build_host_table(
cfg: Config,
services_by_host: dict[str, list[str]],
running_by_host: dict[str, list[str]],
container_counts: dict[str, int],
*,
show_containers: bool,
) -> Table:
"""Build the hosts table."""
table = Table(title="Hosts", show_header=True, header_style="bold cyan")
table.add_column("Host", style="magenta")
table.add_column("Address")
table.add_column("Configured", justify="right")
table.add_column("Running", justify="right")
if show_containers:
table.add_column("Containers", justify="right")
for host_name in sorted(cfg.hosts.keys()):
host = cfg.hosts[host_name]
configured = len(services_by_host[host_name])
running = len(running_by_host[host_name])
row = [
host_name,
host.address,
str(configured),
str(running) if running > 0 else "[dim]0[/]",
]
if show_containers:
count = container_counts.get(host_name, 0)
row.append(str(count) if count > 0 else "[dim]0[/]")
table.add_row(*row)
return table
def _build_summary_table(
cfg: Config, state: dict[str, str | list[str]], pending: list[str]
) -> Table:
"""Build the summary table."""
on_disk = cfg.discover_compose_dirs()
table = Table(title="Summary", show_header=False)
table.add_column("Label", style="dim")
table.add_column("Value", style="bold")
table.add_row("Total hosts", str(len(cfg.hosts)))
table.add_row("Services (configured)", str(len(cfg.services)))
table.add_row("Services (tracked)", str(len(state)))
table.add_row("Compose files on disk", str(len(on_disk)))
if pending:
preview = ", ".join(pending[:_STATS_PREVIEW_LIMIT])
suffix = "..." if len(pending) > _STATS_PREVIEW_LIMIT else ""
table.add_row("Pending migrations", f"[yellow]{len(pending)}[/] ({preview}{suffix})")
else:
table.add_row("Pending migrations", "[green]0[/]")
return table
# --- Command functions ---
@app.command(rich_help_panel="Monitoring")
def logs(
services: ServicesArg = None,
all_services: AllOption = False,
host: HostOption = None,
follow: Annotated[bool, typer.Option("--follow", "-f", help="Follow logs")] = False,
tail: Annotated[
int | None,
typer.Option("--tail", "-n", help="Number of lines (default: 20 for --all, 100 otherwise)"),
] = None,
config: ConfigOption = None,
) -> None:
"""Show service logs."""
if all_services and host is not None:
err_console.print("[red]✗[/] Cannot use --all and --host together")
raise typer.Exit(1)
cfg = load_config_or_exit(config)
# Determine service list based on options
if host is not None:
if host not in cfg.hosts:
err_console.print(f"[red]✗[/] Host '{host}' not found in config")
raise typer.Exit(1)
# Include services where host is in the list of configured hosts
svc_list = [s for s in cfg.services if host in cfg.get_hosts(s)]
if not svc_list:
err_console.print(f"[yellow]![/] No services configured for host '{host}'")
return
else:
svc_list, cfg = get_services(services or [], all_services, config)
# Default to fewer lines when showing multiple services
many_services = all_services or host is not None or len(svc_list) > 1
effective_tail = tail if tail is not None else (20 if many_services else 100)
cmd = f"logs --tail {effective_tail}"
if follow:
cmd += " -f"
results = run_async(run_on_services(cfg, svc_list, cmd))
report_results(results)
@app.command(rich_help_panel="Monitoring")
def ps(
config: ConfigOption = None,
) -> None:
"""Show status of all services."""
cfg = load_config_or_exit(config)
results = run_async(run_on_services(cfg, list(cfg.services.keys()), "ps"))
report_results(results)
@app.command(rich_help_panel="Monitoring")
def stats(
live: Annotated[
bool,
typer.Option("--live", "-l", help="Query Docker for live container stats"),
] = False,
config: ConfigOption = None,
) -> None:
"""Show overview statistics for hosts and services.
Without --live: Shows config/state info (hosts, services, pending migrations).
With --live: Also queries Docker on each host for container counts.
"""
cfg = load_config_or_exit(config)
state = load_state(cfg)
pending = get_services_needing_migration(cfg)
all_hosts = list(cfg.hosts.keys())
services_by_host = _group_services_by_host(cfg.services, cfg.hosts, all_hosts)
running_by_host = _group_services_by_host(state, cfg.hosts, all_hosts)
container_counts: dict[str, int] = {}
if live:
container_counts = _get_container_counts(cfg)
host_table = _build_host_table(
cfg, services_by_host, running_by_host, container_counts, show_containers=live
)
console.print(host_table)
console.print()
console.print(_build_summary_table(cfg, state, pending))

View File

@@ -203,6 +203,51 @@ def parse_host_volumes(config: Config, service: str) -> list[str]:
return unique
def parse_devices(config: Config, service: str) -> list[str]:
"""Extract host device paths from a service's compose file.
Returns a list of host device paths (e.g., /dev/dri, /dev/dri/renderD128).
"""
compose_path = config.get_compose_path(service)
if not compose_path.exists():
return []
env = _load_env(compose_path)
compose_data = yaml.safe_load(compose_path.read_text()) or {}
raw_services = compose_data.get("services", {})
if not isinstance(raw_services, dict):
return []
devices: list[str] = []
for definition in raw_services.values():
if not isinstance(definition, dict):
continue
device_list = definition.get("devices")
if not device_list or not isinstance(device_list, list):
continue
for item in device_list:
if not isinstance(item, str):
continue
interpolated = _interpolate(item, env)
# Format: host_path:container_path[:options]
parts = interpolated.split(":")
if parts:
host_path = parts[0]
if host_path.startswith("/dev/"):
devices.append(host_path)
# Return unique devices, preserving order
seen: set[str] = set()
unique: list[str] = []
for d in devices:
if d not in seen:
seen.add(d)
unique.append(d)
return unique
def parse_external_networks(config: Config, service: str) -> list[str]:
"""Extract external network names from a service's compose file.

View File

@@ -9,10 +9,7 @@ from pathlib import Path
import yaml
from pydantic import BaseModel, Field, model_validator
def xdg_config_home() -> Path:
"""Get XDG config directory, respecting XDG_CONFIG_HOME env var."""
return Path(os.environ.get("XDG_CONFIG_HOME", Path.home() / ".config"))
from .paths import xdg_config_home
class Host(BaseModel):
@@ -148,9 +145,10 @@ def load_config(path: Path | None = None) -> Config:
"""Load configuration from YAML file.
Search order:
1. Explicit path if provided
2. ./compose-farm.yaml
3. $XDG_CONFIG_HOME/compose-farm/compose-farm.yaml (defaults to ~/.config)
1. Explicit path if provided via --config
2. CF_CONFIG environment variable
3. ./compose-farm.yaml
4. $XDG_CONFIG_HOME/compose-farm/compose-farm.yaml (defaults to ~/.config)
"""
search_paths = [
Path("compose-farm.yaml"),
@@ -159,6 +157,8 @@ def load_config(path: Path | None = None) -> Config:
if path:
config_path = path
elif env_path := os.environ.get("CF_CONFIG"):
config_path = Path(env_path)
else:
config_path = None
for p in search_paths:
@@ -170,6 +170,13 @@ def load_config(path: Path | None = None) -> Config:
msg = f"Config file not found. Searched: {', '.join(str(p) for p in search_paths)}"
raise FileNotFoundError(msg)
if config_path.is_dir():
msg = (
f"Config path is a directory, not a file: {config_path}\n"
"This often happens when Docker creates an empty directory for a missing mount."
)
raise FileNotFoundError(msg)
with config_path.open() as f:
raw = yaml.safe_load(f)

View File

@@ -0,0 +1,89 @@
# Compose Farm configuration
# Documentation: https://github.com/basnijholt/compose-farm
#
# This file configures compose-farm to manage Docker Compose services
# across multiple hosts via SSH.
#
# Place this file at:
# - ./compose-farm.yaml (current directory)
# - ~/.config/compose-farm/compose-farm.yaml
# - Or specify with: cf --config /path/to/config.yaml
# - Or set CF_CONFIG environment variable
# ------------------------------------------------------------------------------
# compose_dir: Directory containing service subdirectories with compose files
# ------------------------------------------------------------------------------
# Each subdirectory should contain a compose.yaml (or docker-compose.yml).
# This path must be the same on all hosts (NFS mount recommended).
#
compose_dir: /opt/compose
# ------------------------------------------------------------------------------
# hosts: SSH connection details for each host
# ------------------------------------------------------------------------------
# Simple form:
# hostname: ip-or-fqdn
#
# Full form:
# hostname:
# address: ip-or-fqdn
# user: ssh-username # default: current user
# port: 22 # default: 22
#
# Note: "all" is a reserved keyword and cannot be used as a host name.
#
hosts:
# Example: simple form (uses current user, port 22)
server1: 192.168.1.10
# Example: full form with explicit user
server2:
address: 192.168.1.20
user: admin
# Example: full form with custom port
server3:
address: 192.168.1.30
user: root
port: 2222
# ------------------------------------------------------------------------------
# services: Map service names to their target host(s)
# ------------------------------------------------------------------------------
# Each service name must match a subdirectory in compose_dir.
#
# Single host:
# service-name: hostname
#
# Multiple hosts (explicit list):
# service-name: [host1, host2]
#
# All hosts:
# service-name: all
#
services:
# Example: service runs on a single host
nginx: server1
postgres: server2
# Example: service runs on multiple specific hosts
# prometheus: [server1, server2]
# Example: service runs on ALL hosts (e.g., monitoring agents)
# node-exporter: all
# ------------------------------------------------------------------------------
# traefik_file: (optional) Auto-generate Traefik file-provider config
# ------------------------------------------------------------------------------
# When set, compose-farm automatically regenerates this file after
# up/down/restart/update commands. Traefik watches this file for changes.
#
# traefik_file: /opt/compose/traefik/dynamic.d/compose-farm.yml
# ------------------------------------------------------------------------------
# traefik_service: (optional) Service name running Traefik
# ------------------------------------------------------------------------------
# When generating traefik_file, services on the same host as Traefik are
# skipped (they're handled by Traefik's Docker provider directly).
#
# traefik_service: traefik

View File

@@ -9,7 +9,6 @@ from dataclasses import dataclass
from functools import lru_cache
from typing import TYPE_CHECKING, Any
import asyncssh
from rich.markup import escape
from .console import console, err_console
@@ -53,6 +52,15 @@ class CommandResult:
stdout: str = ""
stderr: str = ""
# SSH returns 255 when connection is closed unexpectedly (e.g., Ctrl+C)
_SSH_CONNECTION_CLOSED = 255
@property
def interrupted(self) -> bool:
"""Check if command was killed by SIGINT (Ctrl+C)."""
# Negative exit codes indicate signal termination; -2 = SIGINT
return self.exit_code < 0 or self.exit_code == self._SSH_CONNECTION_CLOSED
def is_local(host: Host) -> bool:
"""Check if host should run locally (no SSH)."""
@@ -156,6 +164,8 @@ async def _run_ssh_command(
success=result.returncode == 0,
)
import asyncssh # noqa: PLC0415 - lazy import for faster CLI startup
proc: asyncssh.SSHClientProcess[Any]
try:
async with asyncssh.connect( # noqa: SIM117 - conn needed before create_process
@@ -418,12 +428,15 @@ async def check_paths_exist(
"""Check if multiple paths exist on a specific host.
Returns a dict mapping path -> exists.
Handles permission denied as "exists" (path is there, just not accessible).
"""
# Only report missing if stat says "No such file", otherwise assume exists
# (handles permission denied correctly - path exists, just not accessible)
return await _batch_check_existence(
config,
host_name,
paths,
lambda esc: f"test -e '{esc}' && echo 'Y:{esc}' || echo 'N:{esc}'",
lambda esc: f"stat '{esc}' 2>&1 | grep -q 'No such file' && echo 'N:{esc}' || echo 'Y:{esc}'",
"mount-check",
)

View File

@@ -8,8 +8,8 @@ from dataclasses import dataclass
from datetime import UTC, datetime
from typing import TYPE_CHECKING, Any
from .config import xdg_config_home
from .executor import run_compose
from .paths import xdg_config_home
if TYPE_CHECKING:
from collections.abc import Awaitable, Callable, Iterable

View File

@@ -6,24 +6,69 @@ CLI commands are thin wrappers around these functions.
from __future__ import annotations
from typing import TYPE_CHECKING
import asyncio
from typing import TYPE_CHECKING, NamedTuple
from .compose import parse_external_networks, parse_host_volumes
from .compose import parse_devices, parse_external_networks, parse_host_volumes
from .console import console, err_console
from .executor import (
CommandResult,
check_networks_exist,
check_paths_exist,
check_service_running,
run_command,
run_compose,
run_compose_on_host,
)
from .state import get_service_host, set_multi_host_service, set_service_host
from .state import (
get_orphaned_services,
get_service_host,
remove_service,
set_multi_host_service,
set_service_host,
)
if TYPE_CHECKING:
from .config import Config
class OperationInterruptedError(Exception):
"""Raised when a command is interrupted by Ctrl+C."""
class PreflightResult(NamedTuple):
"""Result of pre-flight checks for a service on a host."""
missing_paths: list[str]
missing_networks: list[str]
missing_devices: list[str]
@property
def ok(self) -> bool:
"""Return True if all checks passed."""
return not (self.missing_paths or self.missing_networks or self.missing_devices)
async def _run_compose_step(
cfg: Config,
service: str,
command: str,
*,
raw: bool,
host: str | None = None,
) -> CommandResult:
"""Run a compose command, handle raw output newline, and check for interrupts."""
if host:
result = await run_compose_on_host(cfg, service, host, command, raw=raw)
else:
result = await run_compose(cfg, service, command, raw=raw)
if raw:
print() # Ensure newline after raw output
if result.interrupted:
raise OperationInterruptedError
return result
def get_service_paths(cfg: Config, service: str) -> list[str]:
"""Get all required paths for a service (compose_dir + volumes)."""
paths = [str(cfg.compose_dir)]
@@ -31,58 +76,84 @@ def get_service_paths(cfg: Config, service: str) -> list[str]:
return paths
async def _check_mounts_for_migration(
async def check_service_requirements(
cfg: Config,
service: str,
target_host: str,
) -> list[str]:
"""Check if mount paths exist on target host. Returns list of missing paths."""
paths = get_service_paths(cfg, service)
exists = await check_paths_exist(cfg, target_host, paths)
return [p for p, found in exists.items() if not found]
host_name: str,
) -> PreflightResult:
"""Check if a service can run on a specific host.
async def _check_networks_for_migration(
cfg: Config,
service: str,
target_host: str,
) -> list[str]:
"""Check if Docker networks exist on target host. Returns list of missing networks."""
networks = parse_external_networks(cfg, service)
if not networks:
return []
exists = await check_networks_exist(cfg, target_host, networks)
return [n for n, found in exists.items() if not found]
async def _preflight_check(
cfg: Config,
service: str,
target_host: str,
) -> tuple[list[str], list[str]]:
"""Run pre-flight checks for a service on target host.
Returns (missing_paths, missing_networks).
Verifies that all required paths (volumes), networks, and devices exist.
"""
missing_paths = await _check_mounts_for_migration(cfg, service, target_host)
missing_networks = await _check_networks_for_migration(cfg, service, target_host)
return missing_paths, missing_networks
# Check mount paths
paths = get_service_paths(cfg, service)
path_exists = await check_paths_exist(cfg, host_name, paths)
missing_paths = [p for p, found in path_exists.items() if not found]
# Check external networks
networks = parse_external_networks(cfg, service)
missing_networks: list[str] = []
if networks:
net_exists = await check_networks_exist(cfg, host_name, networks)
missing_networks = [n for n, found in net_exists.items() if not found]
# Check devices
devices = parse_devices(cfg, service)
missing_devices: list[str] = []
if devices:
dev_exists = await check_paths_exist(cfg, host_name, devices)
missing_devices = [d for d, found in dev_exists.items() if not found]
return PreflightResult(missing_paths, missing_networks, missing_devices)
async def _cleanup_and_rollback(
cfg: Config,
service: str,
target_host: str,
current_host: str,
prefix: str,
*,
was_running: bool,
raw: bool = False,
) -> None:
"""Clean up failed start and attempt rollback to old host if it was running."""
err_console.print(
f"{prefix} [yellow]![/] Cleaning up failed start on [magenta]{target_host}[/]"
)
await run_compose(cfg, service, "down", raw=raw)
if not was_running:
err_console.print(
f"{prefix} [dim]Service was not running on [magenta]{current_host}[/], skipping rollback[/]"
)
return
err_console.print(f"{prefix} [yellow]![/] Rolling back to [magenta]{current_host}[/]...")
rollback_result = await run_compose_on_host(cfg, service, current_host, "up -d", raw=raw)
if rollback_result.success:
console.print(f"{prefix} [green]✓[/] Rollback succeeded on [magenta]{current_host}[/]")
else:
err_console.print(f"{prefix} [red]✗[/] Rollback failed - service is down")
def _report_preflight_failures(
service: str,
target_host: str,
missing_paths: list[str],
missing_networks: list[str],
preflight: PreflightResult,
) -> None:
"""Report pre-flight check failures."""
err_console.print(
f"[cyan]\\[{service}][/] [red]✗[/] Cannot start on [magenta]{target_host}[/]:"
)
for path in missing_paths:
for path in preflight.missing_paths:
err_console.print(f" [red]✗[/] missing path: {path}")
for net in missing_networks:
for net in preflight.missing_networks:
err_console.print(f" [red]✗[/] missing network: {net}")
if preflight.missing_networks:
err_console.print(f" [dim]hint: cf init-network {target_host}[/]")
for dev in preflight.missing_devices:
err_console.print(f" [red]✗[/] missing device: {dev}")
async def _up_multi_host_service(
@@ -100,9 +171,9 @@ async def _up_multi_host_service(
# Pre-flight checks on all hosts
for host_name in host_names:
missing_paths, missing_networks = await _preflight_check(cfg, service, host_name)
if missing_paths or missing_networks:
_report_preflight_failures(service, host_name, missing_paths, missing_networks)
preflight = await check_service_requirements(cfg, service, host_name)
if not preflight.ok:
_report_preflight_failures(service, host_name, preflight)
results.append(
CommandResult(service=f"{service}@{host_name}", exit_code=1, success=False)
)
@@ -130,6 +201,97 @@ async def _up_multi_host_service(
return results
async def _migrate_service(
cfg: Config,
service: str,
current_host: str,
target_host: str,
prefix: str,
*,
raw: bool = False,
) -> CommandResult | None:
"""Migrate a service from current_host to target_host.
Pre-pulls/builds images on target, then stops service on current host.
Returns failure result if migration prep fails, None on success.
"""
console.print(
f"{prefix} Migrating from [magenta]{current_host}[/] → [magenta]{target_host}[/]..."
)
# Prepare images on target host before stopping old service to minimize downtime.
# Pull handles image-based services; build handles Dockerfile-based services.
# --ignore-buildable makes pull skip images that have build: defined.
for cmd, label in [("pull --ignore-buildable", "Pull"), ("build", "Build")]:
result = await _run_compose_step(cfg, service, cmd, raw=raw)
if not result.success:
err_console.print(
f"{prefix} [red]✗[/] {label} failed on [magenta]{target_host}[/], "
"leaving service on current host"
)
return result
# Stop on current host
down_result = await _run_compose_step(cfg, service, "down", raw=raw, host=current_host)
return down_result if not down_result.success else None
async def _up_single_service(
cfg: Config,
service: str,
prefix: str,
*,
raw: bool,
) -> CommandResult:
"""Start a single-host service with migration support."""
target_host = cfg.get_hosts(service)[0]
current_host = get_service_host(cfg, service)
# Pre-flight check: verify paths, networks, and devices exist on target
preflight = await check_service_requirements(cfg, service, target_host)
if not preflight.ok:
_report_preflight_failures(service, target_host, preflight)
return CommandResult(service=service, exit_code=1, success=False)
# If service is deployed elsewhere, migrate it
did_migration = False
was_running = False
if current_host and current_host != target_host:
if current_host in cfg.hosts:
was_running = await check_service_running(cfg, service, current_host)
failure = await _migrate_service(
cfg, service, current_host, target_host, prefix, raw=raw
)
if failure:
return failure
did_migration = True
else:
err_console.print(
f"{prefix} [yellow]![/] was on "
f"[magenta]{current_host}[/] (not in config), skipping down"
)
# Start on target host
console.print(f"{prefix} Starting on [magenta]{target_host}[/]...")
up_result = await _run_compose_step(cfg, service, "up -d", raw=raw)
# Update state on success, or rollback on failure
if up_result.success:
set_service_host(cfg, service, target_host)
elif did_migration and current_host:
await _cleanup_and_rollback(
cfg,
service,
target_host,
current_host,
prefix,
was_running=was_running,
raw=raw,
)
return up_result
async def up_services(
cfg: Config,
services: list[str],
@@ -140,54 +302,16 @@ async def up_services(
results: list[CommandResult] = []
total = len(services)
for idx, service in enumerate(services, 1):
prefix = f"[dim][{idx}/{total}][/] [cyan]\\[{service}][/]"
try:
for idx, service in enumerate(services, 1):
prefix = f"[dim][{idx}/{total}][/] [cyan]\\[{service}][/]"
# Handle multi-host services separately (no migration)
if cfg.is_multi_host(service):
multi_results = await _up_multi_host_service(cfg, service, prefix, raw=raw)
results.extend(multi_results)
continue
target_host = cfg.get_hosts(service)[0]
current_host = get_service_host(cfg, service)
# Pre-flight check: verify paths and networks exist on target
missing_paths, missing_networks = await _preflight_check(cfg, service, target_host)
if missing_paths or missing_networks:
_report_preflight_failures(service, target_host, missing_paths, missing_networks)
results.append(CommandResult(service=service, exit_code=1, success=False))
continue
# If service is deployed elsewhere, migrate it
if current_host and current_host != target_host:
if current_host in cfg.hosts:
console.print(
f"{prefix} Migrating from "
f"[magenta]{current_host}[/] → [magenta]{target_host}[/]..."
)
down_result = await run_compose_on_host(cfg, service, current_host, "down", raw=raw)
if raw:
print() # Ensure newline after raw output
if not down_result.success:
results.append(down_result)
continue
if cfg.is_multi_host(service):
results.extend(await _up_multi_host_service(cfg, service, prefix, raw=raw))
else:
err_console.print(
f"{prefix} [yellow]![/] was on "
f"[magenta]{current_host}[/] (not in config), skipping down"
)
# Start on target host
console.print(f"{prefix} Starting on [magenta]{target_host}[/]...")
up_result = await run_compose(cfg, service, "up -d", raw=raw)
if raw:
print() # Ensure newline after raw output (progress bars end with \r)
results.append(up_result)
# Update state on success
if up_result.success:
set_service_host(cfg, service, target_host)
results.append(await _up_single_service(cfg, service, prefix, raw=raw))
except OperationInterruptedError:
raise KeyboardInterrupt from None
return results
@@ -196,17 +320,95 @@ async def check_host_compatibility(
cfg: Config,
service: str,
) -> dict[str, tuple[int, int, list[str]]]:
"""Check which hosts can run a service based on mount paths.
"""Check which hosts can run a service based on paths, networks, and devices.
Returns dict of host_name -> (found_count, total_count, missing_paths).
Returns dict of host_name -> (found_count, total_count, missing_items).
"""
# Get total requirements count
paths = get_service_paths(cfg, service)
networks = parse_external_networks(cfg, service)
devices = parse_devices(cfg, service)
total = len(paths) + len(networks) + len(devices)
results: dict[str, tuple[int, int, list[str]]] = {}
for host_name in cfg.hosts:
exists = await check_paths_exist(cfg, host_name, paths)
found = sum(1 for v in exists.values() if v)
missing = [p for p, v in exists.items() if not v]
results[host_name] = (found, len(paths), missing)
preflight = await check_service_requirements(cfg, service, host_name)
all_missing = (
preflight.missing_paths + preflight.missing_networks + preflight.missing_devices
)
found = total - len(all_missing)
results[host_name] = (found, total, all_missing)
return results
async def stop_orphaned_services(cfg: Config) -> list[CommandResult]:
"""Stop orphaned services (in state but not in config).
Runs docker compose down on each service on its tracked host(s).
Only removes from state on successful stop.
Returns list of CommandResults for each service@host.
"""
orphaned = get_orphaned_services(cfg)
if not orphaned:
return []
results: list[CommandResult] = []
tasks: list[tuple[str, str, asyncio.Task[CommandResult]]] = []
# Build list of (service, host, task) for all orphaned services
for service, hosts in orphaned.items():
host_list = hosts if isinstance(hosts, list) else [hosts]
for host in host_list:
# Skip hosts no longer in config
if host not in cfg.hosts:
console.print(
f" [yellow]![/] {service}@{host}: host no longer in config, skipping"
)
results.append(
CommandResult(
service=f"{service}@{host}",
exit_code=1,
success=False,
stderr="host no longer in config",
)
)
continue
coro = run_compose_on_host(cfg, service, host, "down")
tasks.append((service, host, asyncio.create_task(coro)))
# Run all down commands in parallel
if tasks:
for service, host, task in tasks:
try:
result = await task
results.append(result)
if result.success:
console.print(f" [green]✓[/] {service}@{host}: stopped")
else:
console.print(f" [red]✗[/] {service}@{host}: {result.stderr or 'failed'}")
except Exception as e:
console.print(f" [red]✗[/] {service}@{host}: {e}")
results.append(
CommandResult(
service=f"{service}@{host}",
exit_code=1,
success=False,
stderr=str(e),
)
)
# Remove from state only for services where ALL hosts succeeded
for service, hosts in orphaned.items():
host_list = hosts if isinstance(hosts, list) else [hosts]
all_succeeded = all(
r.success
for r in results
if r.service.startswith(f"{service}@") or r.service == service
)
if all_succeeded:
remove_service(cfg, service)
return results

21
src/compose_farm/paths.py Normal file
View File

@@ -0,0 +1,21 @@
"""Path utilities - lightweight module with no heavy dependencies."""
from __future__ import annotations
import os
from pathlib import Path
def xdg_config_home() -> Path:
"""Get XDG config directory, respecting XDG_CONFIG_HOME env var."""
return Path(os.environ.get("XDG_CONFIG_HOME", Path.home() / ".config"))
def default_config_path() -> Path:
"""Get the default user config path."""
return xdg_config_home() / "compose-farm" / "compose-farm.yaml"
def config_search_paths() -> list[Path]:
"""Get search paths for config files."""
return [Path("compose-farm.yaml"), default_config_path()]

View File

@@ -142,3 +142,25 @@ def get_services_needing_migration(config: Config) -> list[str]:
if current_host and current_host != configured_host:
needs_migration.append(service)
return needs_migration
def get_orphaned_services(config: Config) -> dict[str, str | list[str]]:
"""Get services that are in state but not in config.
These are services that were previously deployed but have been
removed from the config file (e.g., commented out).
Returns a dict mapping service name to host(s) where it's deployed.
"""
state = load_state(config)
return {service: hosts for service, hosts in state.items() if service not in config.services}
def get_services_not_in_state(config: Config) -> list[str]:
"""Get services that are in config but not in state.
These are services that should be running but aren't tracked
(e.g., newly added to config, or previously stopped as orphans).
"""
state = load_state(config)
return [service for service in config.services if service not in state]

426
tests/test_cli_lifecycle.py Normal file
View File

@@ -0,0 +1,426 @@
"""Tests for CLI lifecycle commands (apply, down --orphaned)."""
from pathlib import Path
from unittest.mock import patch
import pytest
import typer
from compose_farm.cli.lifecycle import apply, down
from compose_farm.config import Config, Host
from compose_farm.executor import CommandResult
def _make_config(tmp_path: Path, services: dict[str, str] | None = None) -> Config:
"""Create a minimal config for testing."""
compose_dir = tmp_path / "compose"
compose_dir.mkdir()
svc_dict = services or {"svc1": "host1", "svc2": "host2"}
for svc in svc_dict:
svc_dir = compose_dir / svc
svc_dir.mkdir()
(svc_dir / "docker-compose.yml").write_text("services: {}\n")
config_path = tmp_path / "compose-farm.yaml"
config_path.write_text("")
return Config(
compose_dir=compose_dir,
hosts={"host1": Host(address="localhost"), "host2": Host(address="localhost")},
services=svc_dict,
config_path=config_path,
)
def _make_result(service: str, success: bool = True) -> CommandResult:
"""Create a command result."""
return CommandResult(
service=service,
exit_code=0 if success else 1,
success=success,
stdout="",
stderr="",
)
class TestApplyCommand:
"""Tests for the apply command."""
def test_apply_nothing_to_do(self, tmp_path: Path, capsys: pytest.CaptureFixture[str]) -> None:
"""When no migrations, orphans, or missing services, prints success message."""
cfg = _make_config(tmp_path)
with (
patch("compose_farm.cli.lifecycle.load_config_or_exit", return_value=cfg),
patch("compose_farm.cli.lifecycle.get_orphaned_services", return_value={}),
patch("compose_farm.cli.lifecycle.get_services_needing_migration", return_value=[]),
patch("compose_farm.cli.lifecycle.get_services_not_in_state", return_value=[]),
):
apply(dry_run=False, no_orphans=False, full=False, config=None)
captured = capsys.readouterr()
assert "Nothing to apply" in captured.out
def test_apply_dry_run_shows_preview(
self, tmp_path: Path, capsys: pytest.CaptureFixture[str]
) -> None:
"""Dry run shows what would be done without executing."""
cfg = _make_config(tmp_path)
with (
patch("compose_farm.cli.lifecycle.load_config_or_exit", return_value=cfg),
patch(
"compose_farm.cli.lifecycle.get_orphaned_services",
return_value={"old-svc": "host1"},
),
patch(
"compose_farm.cli.lifecycle.get_services_needing_migration",
return_value=["svc1"],
),
patch("compose_farm.cli.lifecycle.get_services_not_in_state", return_value=[]),
patch("compose_farm.cli.lifecycle.get_service_host", return_value="host1"),
patch("compose_farm.cli.lifecycle.stop_orphaned_services") as mock_stop,
patch("compose_farm.cli.lifecycle.up_services") as mock_up,
):
apply(dry_run=True, no_orphans=False, full=False, config=None)
captured = capsys.readouterr()
assert "Services to migrate" in captured.out
assert "svc1" in captured.out
assert "Orphaned services to stop" in captured.out
assert "old-svc" in captured.out
assert "dry-run" in captured.out
# Should not have called the actual operations
mock_stop.assert_not_called()
mock_up.assert_not_called()
def test_apply_executes_migrations(self, tmp_path: Path) -> None:
"""Apply runs migrations when services need migration."""
cfg = _make_config(tmp_path)
mock_results = [_make_result("svc1")]
with (
patch("compose_farm.cli.lifecycle.load_config_or_exit", return_value=cfg),
patch("compose_farm.cli.lifecycle.get_orphaned_services", return_value={}),
patch(
"compose_farm.cli.lifecycle.get_services_needing_migration",
return_value=["svc1"],
),
patch("compose_farm.cli.lifecycle.get_services_not_in_state", return_value=[]),
patch("compose_farm.cli.lifecycle.get_service_host", return_value="host1"),
patch(
"compose_farm.cli.lifecycle.run_async",
return_value=mock_results,
),
patch("compose_farm.cli.lifecycle.up_services") as mock_up,
patch("compose_farm.cli.lifecycle.maybe_regenerate_traefik"),
patch("compose_farm.cli.lifecycle.report_results"),
):
apply(dry_run=False, no_orphans=False, full=False, config=None)
mock_up.assert_called_once()
call_args = mock_up.call_args
assert call_args[0][1] == ["svc1"] # services list
def test_apply_executes_orphan_cleanup(self, tmp_path: Path) -> None:
"""Apply stops orphaned services."""
cfg = _make_config(tmp_path)
mock_results = [_make_result("old-svc@host1")]
with (
patch("compose_farm.cli.lifecycle.load_config_or_exit", return_value=cfg),
patch(
"compose_farm.cli.lifecycle.get_orphaned_services",
return_value={"old-svc": "host1"},
),
patch("compose_farm.cli.lifecycle.get_services_needing_migration", return_value=[]),
patch("compose_farm.cli.lifecycle.get_services_not_in_state", return_value=[]),
patch(
"compose_farm.cli.lifecycle.run_async",
return_value=mock_results,
),
patch("compose_farm.cli.lifecycle.stop_orphaned_services") as mock_stop,
patch("compose_farm.cli.lifecycle.report_results"),
):
apply(dry_run=False, no_orphans=False, full=False, config=None)
mock_stop.assert_called_once_with(cfg)
def test_apply_no_orphans_skips_orphan_cleanup(
self, tmp_path: Path, capsys: pytest.CaptureFixture[str]
) -> None:
"""--no-orphans flag skips orphan cleanup."""
cfg = _make_config(tmp_path)
mock_results = [_make_result("svc1")]
with (
patch("compose_farm.cli.lifecycle.load_config_or_exit", return_value=cfg),
patch(
"compose_farm.cli.lifecycle.get_orphaned_services",
return_value={"old-svc": "host1"},
),
patch(
"compose_farm.cli.lifecycle.get_services_needing_migration",
return_value=["svc1"],
),
patch("compose_farm.cli.lifecycle.get_services_not_in_state", return_value=[]),
patch("compose_farm.cli.lifecycle.get_service_host", return_value="host1"),
patch(
"compose_farm.cli.lifecycle.run_async",
return_value=mock_results,
),
patch("compose_farm.cli.lifecycle.up_services") as mock_up,
patch("compose_farm.cli.lifecycle.stop_orphaned_services") as mock_stop,
patch("compose_farm.cli.lifecycle.maybe_regenerate_traefik"),
patch("compose_farm.cli.lifecycle.report_results"),
):
apply(dry_run=False, no_orphans=True, full=False, config=None)
# Should run migrations but not orphan cleanup
mock_up.assert_called_once()
mock_stop.assert_not_called()
# Orphans should not appear in output
captured = capsys.readouterr()
assert "old-svc" not in captured.out
def test_apply_no_orphans_nothing_to_do(
self, tmp_path: Path, capsys: pytest.CaptureFixture[str]
) -> None:
"""--no-orphans with only orphans means nothing to do."""
cfg = _make_config(tmp_path)
with (
patch("compose_farm.cli.lifecycle.load_config_or_exit", return_value=cfg),
patch(
"compose_farm.cli.lifecycle.get_orphaned_services",
return_value={"old-svc": "host1"},
),
patch("compose_farm.cli.lifecycle.get_services_needing_migration", return_value=[]),
patch("compose_farm.cli.lifecycle.get_services_not_in_state", return_value=[]),
):
apply(dry_run=False, no_orphans=True, full=False, config=None)
captured = capsys.readouterr()
assert "Nothing to apply" in captured.out
def test_apply_starts_missing_services(self, tmp_path: Path) -> None:
"""Apply starts services that are in config but not in state."""
cfg = _make_config(tmp_path)
mock_results = [_make_result("svc1")]
with (
patch("compose_farm.cli.lifecycle.load_config_or_exit", return_value=cfg),
patch("compose_farm.cli.lifecycle.get_orphaned_services", return_value={}),
patch("compose_farm.cli.lifecycle.get_services_needing_migration", return_value=[]),
patch(
"compose_farm.cli.lifecycle.get_services_not_in_state",
return_value=["svc1"],
),
patch(
"compose_farm.cli.lifecycle.run_async",
return_value=mock_results,
),
patch("compose_farm.cli.lifecycle.up_services") as mock_up,
patch("compose_farm.cli.lifecycle.maybe_regenerate_traefik"),
patch("compose_farm.cli.lifecycle.report_results"),
):
apply(dry_run=False, no_orphans=False, full=False, config=None)
mock_up.assert_called_once()
call_args = mock_up.call_args
assert call_args[0][1] == ["svc1"]
def test_apply_dry_run_shows_missing_services(
self, tmp_path: Path, capsys: pytest.CaptureFixture[str]
) -> None:
"""Dry run shows services that would be started."""
cfg = _make_config(tmp_path)
with (
patch("compose_farm.cli.lifecycle.load_config_or_exit", return_value=cfg),
patch("compose_farm.cli.lifecycle.get_orphaned_services", return_value={}),
patch("compose_farm.cli.lifecycle.get_services_needing_migration", return_value=[]),
patch(
"compose_farm.cli.lifecycle.get_services_not_in_state",
return_value=["svc1"],
),
):
apply(dry_run=True, no_orphans=False, full=False, config=None)
captured = capsys.readouterr()
assert "Services to start" in captured.out
assert "svc1" in captured.out
assert "dry-run" in captured.out
def test_apply_full_refreshes_all_services(self, tmp_path: Path) -> None:
"""--full runs up on all services to pick up config changes."""
cfg = _make_config(tmp_path)
mock_results = [_make_result("svc1"), _make_result("svc2")]
with (
patch("compose_farm.cli.lifecycle.load_config_or_exit", return_value=cfg),
patch("compose_farm.cli.lifecycle.get_orphaned_services", return_value={}),
patch("compose_farm.cli.lifecycle.get_services_needing_migration", return_value=[]),
patch("compose_farm.cli.lifecycle.get_services_not_in_state", return_value=[]),
patch(
"compose_farm.cli.lifecycle.run_async",
return_value=mock_results,
),
patch("compose_farm.cli.lifecycle.up_services") as mock_up,
patch("compose_farm.cli.lifecycle.maybe_regenerate_traefik"),
patch("compose_farm.cli.lifecycle.report_results"),
):
apply(dry_run=False, no_orphans=False, full=True, config=None)
mock_up.assert_called_once()
call_args = mock_up.call_args
# Should refresh all services in config
assert set(call_args[0][1]) == {"svc1", "svc2"}
def test_apply_full_dry_run_shows_refresh(
self, tmp_path: Path, capsys: pytest.CaptureFixture[str]
) -> None:
"""--full --dry-run shows services that would be refreshed."""
cfg = _make_config(tmp_path)
with (
patch("compose_farm.cli.lifecycle.load_config_or_exit", return_value=cfg),
patch("compose_farm.cli.lifecycle.get_orphaned_services", return_value={}),
patch("compose_farm.cli.lifecycle.get_services_needing_migration", return_value=[]),
patch("compose_farm.cli.lifecycle.get_services_not_in_state", return_value=[]),
):
apply(dry_run=True, no_orphans=False, full=True, config=None)
captured = capsys.readouterr()
assert "Services to refresh" in captured.out
assert "svc1" in captured.out
assert "svc2" in captured.out
assert "dry-run" in captured.out
def test_apply_full_excludes_already_handled_services(self, tmp_path: Path) -> None:
"""--full doesn't double-process services that are migrating or starting."""
cfg = _make_config(tmp_path, {"svc1": "host1", "svc2": "host2", "svc3": "host1"})
mock_results = [_make_result("svc1"), _make_result("svc3")]
with (
patch("compose_farm.cli.lifecycle.load_config_or_exit", return_value=cfg),
patch("compose_farm.cli.lifecycle.get_orphaned_services", return_value={}),
patch(
"compose_farm.cli.lifecycle.get_services_needing_migration",
return_value=["svc1"],
),
patch(
"compose_farm.cli.lifecycle.get_services_not_in_state",
return_value=["svc2"],
),
patch("compose_farm.cli.lifecycle.get_service_host", return_value="host2"),
patch(
"compose_farm.cli.lifecycle.run_async",
return_value=mock_results,
),
patch("compose_farm.cli.lifecycle.up_services") as mock_up,
patch("compose_farm.cli.lifecycle.maybe_regenerate_traefik"),
patch("compose_farm.cli.lifecycle.report_results"),
):
apply(dry_run=False, no_orphans=False, full=True, config=None)
# up_services should be called 3 times: migrate, start, refresh
assert mock_up.call_count == 3
# Get the third call (refresh) and check it only has svc3
refresh_call = mock_up.call_args_list[2]
assert refresh_call[0][1] == ["svc3"]
class TestDownOrphaned:
"""Tests for down --orphaned flag."""
def test_down_orphaned_no_orphans(
self, tmp_path: Path, capsys: pytest.CaptureFixture[str]
) -> None:
"""When no orphans exist, prints success message."""
cfg = _make_config(tmp_path)
with (
patch("compose_farm.cli.lifecycle.load_config_or_exit", return_value=cfg),
patch("compose_farm.cli.lifecycle.get_orphaned_services", return_value={}),
):
down(
services=None,
all_services=False,
orphaned=True,
host=None,
config=None,
)
captured = capsys.readouterr()
assert "No orphaned services to stop" in captured.out
def test_down_orphaned_stops_services(self, tmp_path: Path) -> None:
"""--orphaned stops orphaned services."""
cfg = _make_config(tmp_path)
mock_results = [_make_result("old-svc@host1")]
with (
patch("compose_farm.cli.lifecycle.load_config_or_exit", return_value=cfg),
patch(
"compose_farm.cli.lifecycle.get_orphaned_services",
return_value={"old-svc": "host1"},
),
patch(
"compose_farm.cli.lifecycle.run_async",
return_value=mock_results,
),
patch("compose_farm.cli.lifecycle.stop_orphaned_services") as mock_stop,
patch("compose_farm.cli.lifecycle.report_results"),
):
down(
services=None,
all_services=False,
orphaned=True,
host=None,
config=None,
)
mock_stop.assert_called_once_with(cfg)
def test_down_orphaned_with_services_errors(self) -> None:
"""--orphaned cannot be combined with service arguments."""
with pytest.raises(typer.Exit) as exc_info:
down(
services=["svc1"],
all_services=False,
orphaned=True,
host=None,
config=None,
)
assert exc_info.value.exit_code == 1
def test_down_orphaned_with_all_errors(self) -> None:
"""--orphaned cannot be combined with --all."""
with pytest.raises(typer.Exit) as exc_info:
down(
services=None,
all_services=True,
orphaned=True,
host=None,
config=None,
)
assert exc_info.value.exit_code == 1
def test_down_orphaned_with_host_errors(self) -> None:
"""--orphaned cannot be combined with --host."""
with pytest.raises(typer.Exit) as exc_info:
down(
services=None,
all_services=False,
orphaned=True,
host="host1",
config=None,
)
assert exc_info.value.exit_code == 1

View File

@@ -8,7 +8,7 @@ from unittest.mock import patch
import pytest
import typer
from compose_farm.cli import logs
from compose_farm.cli.monitoring import logs
from compose_farm.config import Config, Host
from compose_farm.executor import CommandResult
@@ -37,7 +37,7 @@ def _make_result(service: str) -> CommandResult:
def _mock_run_async_factory(
services: list[str],
) -> tuple[Any, list[CommandResult]]:
"""Create a mock _run_async that returns results for given services."""
"""Create a mock run_async that returns results for given services."""
results = [_make_result(s) for s in services]
def mock_run_async(_coro: Coroutine[Any, Any, Any]) -> list[CommandResult]:
@@ -55,9 +55,10 @@ class TestLogsContextualDefault:
mock_run_async, _ = _mock_run_async_factory(["svc1", "svc2", "svc3"])
with (
patch("compose_farm.cli._load_config_or_exit", return_value=cfg),
patch("compose_farm.cli._run_async", side_effect=mock_run_async),
patch("compose_farm.cli.run_on_services") as mock_run,
patch("compose_farm.cli.monitoring.load_config_or_exit", return_value=cfg),
patch("compose_farm.cli.common.load_config_or_exit", return_value=cfg),
patch("compose_farm.cli.monitoring.run_async", side_effect=mock_run_async),
patch("compose_farm.cli.monitoring.run_on_services") as mock_run,
):
mock_run.return_value = None
@@ -73,9 +74,10 @@ class TestLogsContextualDefault:
mock_run_async, _ = _mock_run_async_factory(["svc1"])
with (
patch("compose_farm.cli._load_config_or_exit", return_value=cfg),
patch("compose_farm.cli._run_async", side_effect=mock_run_async),
patch("compose_farm.cli.run_on_services") as mock_run,
patch("compose_farm.cli.monitoring.load_config_or_exit", return_value=cfg),
patch("compose_farm.cli.common.load_config_or_exit", return_value=cfg),
patch("compose_farm.cli.monitoring.run_async", side_effect=mock_run_async),
patch("compose_farm.cli.monitoring.run_on_services") as mock_run,
):
logs(
services=["svc1"],
@@ -96,9 +98,10 @@ class TestLogsContextualDefault:
mock_run_async, _ = _mock_run_async_factory(["svc1", "svc2", "svc3"])
with (
patch("compose_farm.cli._load_config_or_exit", return_value=cfg),
patch("compose_farm.cli._run_async", side_effect=mock_run_async),
patch("compose_farm.cli.run_on_services") as mock_run,
patch("compose_farm.cli.monitoring.load_config_or_exit", return_value=cfg),
patch("compose_farm.cli.common.load_config_or_exit", return_value=cfg),
patch("compose_farm.cli.monitoring.run_async", side_effect=mock_run_async),
patch("compose_farm.cli.monitoring.run_on_services") as mock_run,
):
logs(
services=None,
@@ -119,9 +122,10 @@ class TestLogsContextualDefault:
mock_run_async, _ = _mock_run_async_factory(["svc1"])
with (
patch("compose_farm.cli._load_config_or_exit", return_value=cfg),
patch("compose_farm.cli._run_async", side_effect=mock_run_async),
patch("compose_farm.cli.run_on_services") as mock_run,
patch("compose_farm.cli.monitoring.load_config_or_exit", return_value=cfg),
patch("compose_farm.cli.common.load_config_or_exit", return_value=cfg),
patch("compose_farm.cli.monitoring.run_async", side_effect=mock_run_async),
patch("compose_farm.cli.monitoring.run_on_services") as mock_run,
):
logs(
services=["svc1"],
@@ -146,9 +150,9 @@ class TestLogsHostFilter:
mock_run_async, _ = _mock_run_async_factory(["svc1", "svc2"])
with (
patch("compose_farm.cli._load_config_or_exit", return_value=cfg),
patch("compose_farm.cli._run_async", side_effect=mock_run_async),
patch("compose_farm.cli.run_on_services") as mock_run,
patch("compose_farm.cli.monitoring.load_config_or_exit", return_value=cfg),
patch("compose_farm.cli.monitoring.run_async", side_effect=mock_run_async),
patch("compose_farm.cli.monitoring.run_on_services") as mock_run,
):
logs(
services=None,
@@ -170,9 +174,9 @@ class TestLogsHostFilter:
mock_run_async, _ = _mock_run_async_factory(["svc1", "svc2"])
with (
patch("compose_farm.cli._load_config_or_exit", return_value=cfg),
patch("compose_farm.cli._run_async", side_effect=mock_run_async),
patch("compose_farm.cli.run_on_services") as mock_run,
patch("compose_farm.cli.monitoring.load_config_or_exit", return_value=cfg),
patch("compose_farm.cli.monitoring.run_async", side_effect=mock_run_async),
patch("compose_farm.cli.monitoring.run_on_services") as mock_run,
):
logs(
services=None,
@@ -187,14 +191,10 @@ class TestLogsHostFilter:
call_args = mock_run.call_args
assert call_args[0][2] == "logs --tail 20"
def test_logs_all_and_host_mutually_exclusive(self, tmp_path: Path) -> None:
def test_logs_all_and_host_mutually_exclusive(self) -> None:
"""Using --all and --host together should error."""
cfg = _make_config(tmp_path)
with (
patch("compose_farm.cli._load_config_or_exit", return_value=cfg),
pytest.raises(typer.Exit) as exc_info,
):
# No config mock needed - error is raised before config is loaded
with pytest.raises(typer.Exit) as exc_info:
logs(
services=None,
all_services=True,

View File

@@ -128,6 +128,8 @@ class TestLoadConfig:
def test_load_config_not_found(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.chdir(tmp_path)
monkeypatch.delenv("CF_CONFIG", raising=False)
monkeypatch.setenv("XDG_CONFIG_HOME", str(tmp_path / "empty_config"))
with pytest.raises(FileNotFoundError, match="Config file not found"):
load_config()

230
tests/test_config_cmd.py Normal file
View File

@@ -0,0 +1,230 @@
"""Tests for config command module."""
from pathlib import Path
from typing import Any
import pytest
import yaml
from typer.testing import CliRunner
from compose_farm.cli import app
from compose_farm.cli.config import (
_generate_template,
_get_config_file,
_get_editor,
)
@pytest.fixture
def runner() -> CliRunner:
return CliRunner()
@pytest.fixture
def valid_config_data() -> dict[str, Any]:
return {
"compose_dir": "/opt/compose",
"hosts": {"server1": "192.168.1.10"},
"services": {"nginx": "server1"},
}
class TestGetEditor:
"""Tests for _get_editor function."""
def test_uses_editor_env(self, monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setenv("EDITOR", "code")
monkeypatch.delenv("VISUAL", raising=False)
assert _get_editor() == "code"
def test_uses_visual_env(self, monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.delenv("EDITOR", raising=False)
monkeypatch.setenv("VISUAL", "subl")
assert _get_editor() == "subl"
def test_editor_takes_precedence(self, monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setenv("EDITOR", "vim")
monkeypatch.setenv("VISUAL", "code")
assert _get_editor() == "vim"
class TestGetConfigFile:
"""Tests for _get_config_file function."""
def test_explicit_path(self, tmp_path: Path) -> None:
config_file = tmp_path / "my-config.yaml"
config_file.touch()
result = _get_config_file(config_file)
assert result == config_file.resolve()
def test_cf_config_env(self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
config_file = tmp_path / "env-config.yaml"
config_file.touch()
monkeypatch.setenv("CF_CONFIG", str(config_file))
result = _get_config_file(None)
assert result == config_file.resolve()
def test_returns_none_when_not_found(
self, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.chdir(tmp_path)
monkeypatch.delenv("CF_CONFIG", raising=False)
# Set XDG_CONFIG_HOME to a nonexistent path - config_search_paths() will
# now return paths that don't exist
monkeypatch.setenv("XDG_CONFIG_HOME", str(tmp_path / "nonexistent"))
result = _get_config_file(None)
assert result is None
class TestGenerateTemplate:
"""Tests for _generate_template function."""
def test_generates_valid_yaml(self) -> None:
template = _generate_template()
# Should be valid YAML
data = yaml.safe_load(template)
assert "compose_dir" in data
assert "hosts" in data
assert "services" in data
def test_has_documentation_comments(self) -> None:
template = _generate_template()
assert "# Compose Farm configuration" in template
assert "hosts:" in template
assert "services:" in template
class TestConfigInit:
"""Tests for cf config init command."""
def test_init_creates_file(
self, runner: CliRunner, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.delenv("CF_CONFIG", raising=False)
config_file = tmp_path / "new-config.yaml"
result = runner.invoke(app, ["config", "init", "-p", str(config_file)])
assert result.exit_code == 0
assert config_file.exists()
assert "Config file created" in result.stdout
def test_init_force_overwrites(
self, runner: CliRunner, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.delenv("CF_CONFIG", raising=False)
config_file = tmp_path / "existing.yaml"
config_file.write_text("old content")
result = runner.invoke(app, ["config", "init", "-p", str(config_file), "-f"])
assert result.exit_code == 0
content = config_file.read_text()
assert "old content" not in content
assert "compose_dir" in content
def test_init_prompts_on_existing(
self, runner: CliRunner, tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.delenv("CF_CONFIG", raising=False)
config_file = tmp_path / "existing.yaml"
config_file.write_text("old content")
result = runner.invoke(app, ["config", "init", "-p", str(config_file)], input="n\n")
assert result.exit_code == 0
assert "Aborted" in result.stdout
assert config_file.read_text() == "old content"
class TestConfigPath:
"""Tests for cf config path command."""
def test_path_shows_config(
self,
runner: CliRunner,
tmp_path: Path,
valid_config_data: dict[str, Any],
monkeypatch: pytest.MonkeyPatch,
) -> None:
monkeypatch.chdir(tmp_path)
monkeypatch.delenv("CF_CONFIG", raising=False)
config_file = tmp_path / "compose-farm.yaml"
config_file.write_text(yaml.dump(valid_config_data))
result = runner.invoke(app, ["config", "path"])
assert result.exit_code == 0
assert str(config_file) in result.stdout
def test_path_with_explicit_path(self, runner: CliRunner, tmp_path: Path) -> None:
# When explicitly provided, path is returned even if file doesn't exist
nonexistent = tmp_path / "nonexistent.yaml"
result = runner.invoke(app, ["config", "path", "-p", str(nonexistent)])
assert result.exit_code == 0
assert str(nonexistent) in result.stdout
class TestConfigShow:
"""Tests for cf config show command."""
def test_show_displays_content(
self,
runner: CliRunner,
tmp_path: Path,
valid_config_data: dict[str, Any],
monkeypatch: pytest.MonkeyPatch,
) -> None:
monkeypatch.chdir(tmp_path)
monkeypatch.delenv("CF_CONFIG", raising=False)
config_file = tmp_path / "compose-farm.yaml"
config_file.write_text(yaml.dump(valid_config_data))
result = runner.invoke(app, ["config", "show"])
assert result.exit_code == 0
assert "Config file:" in result.stdout
def test_show_raw_output(
self,
runner: CliRunner,
tmp_path: Path,
valid_config_data: dict[str, Any],
monkeypatch: pytest.MonkeyPatch,
) -> None:
monkeypatch.chdir(tmp_path)
monkeypatch.delenv("CF_CONFIG", raising=False)
config_file = tmp_path / "compose-farm.yaml"
content = yaml.dump(valid_config_data)
config_file.write_text(content)
result = runner.invoke(app, ["config", "show", "-r"])
assert result.exit_code == 0
assert content in result.stdout
class TestConfigValidate:
"""Tests for cf config validate command."""
def test_validate_valid_config(
self,
runner: CliRunner,
tmp_path: Path,
valid_config_data: dict[str, Any],
monkeypatch: pytest.MonkeyPatch,
) -> None:
monkeypatch.chdir(tmp_path)
monkeypatch.delenv("CF_CONFIG", raising=False)
config_file = tmp_path / "compose-farm.yaml"
config_file.write_text(yaml.dump(valid_config_data))
result = runner.invoke(app, ["config", "validate"])
assert result.exit_code == 0
assert "Valid config" in result.stdout
assert "Hosts: 1" in result.stdout
assert "Services: 1" in result.stdout
def test_validate_invalid_config(self, runner: CliRunner, tmp_path: Path) -> None:
config_file = tmp_path / "invalid.yaml"
config_file.write_text("invalid: [yaml: content")
result = runner.invoke(app, ["config", "validate", "-p", str(config_file)])
assert result.exit_code == 1
# Error goes to stderr (captured in output when using CliRunner)
output = result.stdout + (result.stderr or "")
assert "Invalid config" in output or "" in output
def test_validate_missing_config(self, runner: CliRunner, tmp_path: Path) -> None:
nonexistent = tmp_path / "nonexistent.yaml"
result = runner.invoke(app, ["config", "validate", "-p", str(nonexistent)])
assert result.exit_code == 1
# Error goes to stderr
output = result.stdout + (result.stderr or "")
assert "Config file not found" in output or "not found" in output.lower()

111
tests/test_operations.py Normal file
View File

@@ -0,0 +1,111 @@
"""Tests for operations module."""
from __future__ import annotations
import inspect
from pathlib import Path # noqa: TC003
from unittest.mock import patch
import pytest
from compose_farm.cli import lifecycle
from compose_farm.config import Config, Host
from compose_farm.executor import CommandResult
from compose_farm.operations import _migrate_service
@pytest.fixture
def basic_config(tmp_path: Path) -> Config:
"""Create a basic test config."""
compose_dir = tmp_path / "compose"
service_dir = compose_dir / "test-service"
service_dir.mkdir(parents=True)
(service_dir / "docker-compose.yml").write_text("services: {}")
return Config(
compose_dir=compose_dir,
hosts={
"host1": Host(address="localhost"),
"host2": Host(address="localhost"),
},
services={"test-service": "host2"},
)
class TestMigrationCommands:
"""Tests for migration command sequence."""
@pytest.fixture
def config(self, tmp_path: Path) -> Config:
"""Create a test config."""
compose_dir = tmp_path / "compose"
service_dir = compose_dir / "test-service"
service_dir.mkdir(parents=True)
(service_dir / "docker-compose.yml").write_text("services: {}")
return Config(
compose_dir=compose_dir,
hosts={
"host1": Host(address="localhost"),
"host2": Host(address="localhost"),
},
services={"test-service": "host2"},
)
async def test_migration_uses_pull_ignore_buildable(self, config: Config) -> None:
"""Migration should use 'pull --ignore-buildable' to skip buildable images."""
commands_called: list[str] = []
async def mock_run_compose_step(
cfg: Config, # noqa: ARG001
service: str,
command: str,
*,
raw: bool, # noqa: ARG001
host: str | None = None, # noqa: ARG001
) -> CommandResult:
commands_called.append(command)
return CommandResult(
service=service,
exit_code=0,
success=True,
)
with patch(
"compose_farm.operations._run_compose_step",
side_effect=mock_run_compose_step,
):
await _migrate_service(
config,
"test-service",
current_host="host1",
target_host="host2",
prefix="[test]",
raw=False,
)
# Migration should call pull with --ignore-buildable, then build, then down
assert "pull --ignore-buildable" in commands_called
assert "build" in commands_called
assert "down" in commands_called
# pull should come before build
pull_idx = commands_called.index("pull --ignore-buildable")
build_idx = commands_called.index("build")
assert pull_idx < build_idx
class TestUpdateCommandSequence:
"""Tests for update command sequence."""
def test_update_command_sequence_includes_build(self) -> None:
"""Update command should use pull --ignore-buildable and build."""
# This is a static check of the command sequence in lifecycle.py
# The actual command sequence is defined in the update function
source = inspect.getsource(lifecycle.update)
# Verify the command sequence includes pull --ignore-buildable
assert "pull --ignore-buildable" in source
# Verify build is included
assert '"build"' in source or "'build'" in source
# Verify the sequence is pull, build, down, up
assert "down" in source
assert "up -d" in source

View File

@@ -5,9 +5,9 @@ from unittest.mock import AsyncMock, patch
import pytest
from compose_farm import cli as cli_module
from compose_farm import executor as executor_module
from compose_farm import state as state_module
from compose_farm.cli import management as cli_management_module
from compose_farm.config import Config, Host
from compose_farm.executor import CommandResult, check_service_running
@@ -98,7 +98,7 @@ class TestReportSyncChanges:
def test_reports_added(self, capsys: pytest.CaptureFixture[str]) -> None:
"""Reports newly discovered services."""
cli_module._report_sync_changes(
cli_management_module._report_sync_changes(
added=["plex", "jellyfin"],
removed=[],
changed=[],
@@ -112,7 +112,7 @@ class TestReportSyncChanges:
def test_reports_removed(self, capsys: pytest.CaptureFixture[str]) -> None:
"""Reports services that are no longer running."""
cli_module._report_sync_changes(
cli_management_module._report_sync_changes(
added=[],
removed=["sonarr"],
changed=[],
@@ -125,7 +125,7 @@ class TestReportSyncChanges:
def test_reports_changed(self, capsys: pytest.CaptureFixture[str]) -> None:
"""Reports services that moved to a different host."""
cli_module._report_sync_changes(
cli_management_module._report_sync_changes(
added=[],
removed=[],
changed=[("plex", "nas01", "nas02")],

View File

@@ -6,7 +6,9 @@ import pytest
from compose_farm.config import Config, Host
from compose_farm.state import (
get_orphaned_services,
get_service_host,
get_services_not_in_state,
load_state,
remove_service,
save_state,
@@ -130,3 +132,110 @@ class TestRemoveService:
result = load_state(config)
assert result["plex"] == "nas01"
class TestGetOrphanedServices:
"""Tests for get_orphaned_services function."""
def test_no_orphans(self, config: Config) -> None:
"""Returns empty dict when all services in state are in config."""
state_file = config.get_state_path()
state_file.write_text("deployed:\n plex: nas01\n")
result = get_orphaned_services(config)
assert result == {}
def test_finds_orphaned_service(self, config: Config) -> None:
"""Returns services in state but not in config."""
state_file = config.get_state_path()
state_file.write_text("deployed:\n plex: nas01\n jellyfin: nas02\n")
result = get_orphaned_services(config)
# plex is in config, jellyfin is not
assert result == {"jellyfin": "nas02"}
def test_finds_orphaned_multi_host_service(self, config: Config) -> None:
"""Returns multi-host orphaned services with host list."""
state_file = config.get_state_path()
state_file.write_text("deployed:\n plex: nas01\n dozzle:\n - nas01\n - nas02\n")
result = get_orphaned_services(config)
assert result == {"dozzle": ["nas01", "nas02"]}
def test_empty_state(self, config: Config) -> None:
"""Returns empty dict when state is empty."""
result = get_orphaned_services(config)
assert result == {}
def test_all_orphaned(self, tmp_path: Path) -> None:
"""Returns all services when none are in config."""
config_path = tmp_path / "compose-farm.yaml"
config_path.write_text("")
cfg = Config(
compose_dir=tmp_path / "compose",
hosts={"nas01": Host(address="192.168.1.10")},
services={}, # No services in config
config_path=config_path,
)
state_file = cfg.get_state_path()
state_file.write_text("deployed:\n plex: nas01\n jellyfin: nas02\n")
result = get_orphaned_services(cfg)
assert result == {"plex": "nas01", "jellyfin": "nas02"}
class TestGetServicesNotInState:
"""Tests for get_services_not_in_state function."""
def test_all_in_state(self, config: Config) -> None:
"""Returns empty list when all services are in state."""
state_file = config.get_state_path()
state_file.write_text("deployed:\n plex: nas01\n")
result = get_services_not_in_state(config)
assert result == []
def test_finds_missing_service(self, tmp_path: Path) -> None:
"""Returns services in config but not in state."""
config_path = tmp_path / "compose-farm.yaml"
config_path.write_text("")
cfg = Config(
compose_dir=tmp_path / "compose",
hosts={"nas01": Host(address="192.168.1.10")},
services={"plex": "nas01", "jellyfin": "nas01"},
config_path=config_path,
)
state_file = cfg.get_state_path()
state_file.write_text("deployed:\n plex: nas01\n")
result = get_services_not_in_state(cfg)
assert result == ["jellyfin"]
def test_empty_state(self, tmp_path: Path) -> None:
"""Returns all services when state is empty."""
config_path = tmp_path / "compose-farm.yaml"
config_path.write_text("")
cfg = Config(
compose_dir=tmp_path / "compose",
hosts={"nas01": Host(address="192.168.1.10")},
services={"plex": "nas01", "jellyfin": "nas01"},
config_path=config_path,
)
result = get_services_not_in_state(cfg)
assert set(result) == {"plex", "jellyfin"}
def test_empty_config(self, config: Config) -> None:
"""Returns empty list when config has no services."""
# config fixture has plex: nas01, but we need empty config
config_path = config.config_path
config_path.write_text("")
cfg = Config(
compose_dir=config.compose_dir,
hosts={"nas01": Host(address="192.168.1.10")},
services={},
config_path=config_path,
)
result = get_services_not_in_state(cfg)
assert result == []