yadunut/clan-core
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

Fixing a multitude of tests

Browse Source
This commit is contained in:
Qubasa
2023-10-23 22:31:12 +02:00
parent adffdc14f6
commit 8ff80b025c
25 changed files with 154 additions and 113 deletions
Download Patch File Download Diff File Expand all files Collapse all files

13
pkgs/clan-cli/tests/fixtures_flakes.py
View File

@@ -38,7 +38,7 @@ class FlakeForTest(NamedTuple):
def create_flake(
monkeypatch: pytest.MonkeyPatch,
temporary_dir: Path,
temporary_home: Path,
flake_name: FlakeName,
clan_core_flake: Path | None = None,
machines: list[str] = [],
@@ -51,7 +51,7 @@ def create_flake(
template = Path(__file__).parent / flake_name
# copy the template to a new temporary location
home = Path(temporary_dir)
home = Path(temporary_home)
flake = home / ".local/state/clan/flake" / flake_name
shutil.copytree(template, flake)
@@ -87,21 +87,21 @@ def test_flake(
@pytest.fixture
def test_flake_with_core(
monkeypatch: pytest.MonkeyPatch, temporary_dir: Path
monkeypatch: pytest.MonkeyPatch, temporary_home: Path
) -> Iterator[FlakeForTest]:
if not (CLAN_CORE / "flake.nix").exists():
raise Exception(
"clan-core flake not found. This test requires the clan-core flake to be present"
)
yield from create_flake(
monkeypatch, temporary_dir, FlakeName("test_flake_with_core"), CLAN_CORE
monkeypatch, temporary_home, FlakeName("test_flake_with_core"), CLAN_CORE
)
@pytest.fixture
def test_flake_with_core_and_pass(
monkeypatch: pytest.MonkeyPatch,
temporary_dir: Path,
temporary_home: Path,
) -> Iterator[FlakeForTest]:
if not (CLAN_CORE / "flake.nix").exists():
raise Exception(
@@ -109,7 +109,8 @@ def test_flake_with_core_and_pass(
)
yield from create_flake(
monkeypatch,
temporary_dir,
temporary_home,
FlakeName("test_flake_with_core_and_pass"),
CLAN_CORE,
)

10
pkgs/clan-cli/tests/test_dirs.py
View File

@@ -7,15 +7,15 @@ from clan_cli.errors import ClanError
def test_get_clan_flake_toplevel(
monkeypatch: pytest.MonkeyPatch, temporary_dir: Path
monkeypatch: pytest.MonkeyPatch, temporary_home: Path
) -> None:
monkeypatch.chdir(temporary_dir)
monkeypatch.chdir(temporary_home)
with pytest.raises(ClanError):
print(_get_clan_flake_toplevel())
(temporary_dir / ".git").touch()
assert _get_clan_flake_toplevel() == temporary_dir
(temporary_home / ".git").touch()
assert _get_clan_flake_toplevel() == temporary_home
subdir = temporary_dir / "subdir"
subdir = temporary_home / "subdir"
subdir.mkdir()
monkeypatch.chdir(subdir)
(subdir / ".clan-flake").touch()

10
pkgs/clan-cli/tests/test_flake_api.py
View File

@@ -1,13 +1,13 @@
import json
from pathlib import Path
from fixtures_flakes import FlakeForTest
import pytest
from api import TestClient
@pytest.mark.impure
def test_inspect_ok(api: TestClient, test_flake_with_core: Path) -> None:
params = {"url": str(test_flake_with_core)}
def test_inspect_ok(api: TestClient, test_flake_with_core: FlakeForTest) -> None:
params = {"url": str(test_flake_with_core.path)}
response = api.get(
"/api/flake/attrs",
params=params,
@@ -32,8 +32,8 @@ def test_inspect_err(api: TestClient) -> None:
@pytest.mark.impure
def test_inspect_flake(api: TestClient, test_flake_with_core: Path) -> None:
params = {"url": str(test_flake_with_core)}
def test_inspect_flake(api: TestClient, test_flake_with_core: FlakeForTest) -> None:
params = {"url": str(test_flake_with_core.path)}
response = api.get(
"/api/flake",
params=params,

2
pkgs/clan-cli/tests/test_flake_with_core/flake.nix
View File

@@ -9,7 +9,7 @@
let
clan = clan-core.lib.buildClan {
directory = self;
clanName = "test_with_core_clan";
clanName = "test_flake_with_core";
machines = {
vm1 = { lib, ... }: {
clan.networking.deploymentAddress = "__CLAN_DEPLOYMENT_ADDRESS__";

2
pkgs/clan-cli/tests/test_flake_with_core_and_pass/flake.nix
View File

@@ -9,7 +9,7 @@
let
clan = clan-core.lib.buildClan {
directory = self;
clanName = "test_with_core_and_pass_clan";
clanName = "test_flake_with_core_and_pass";
machines = {
vm1 = { lib, ... }: {
clan.networking.deploymentAddress = "__CLAN_DEPLOYMENT_ADDRESS__";

2
pkgs/clan-cli/tests/test_flake_with_core_dynamic_machines/flake.nix
View File

@@ -9,7 +9,7 @@
let
clan = clan-core.lib.buildClan {
directory = self;
clanName = "core_dynamic_machine_clan";
clanName = "test_flake_with_core_dynamic_machines";
machines =
let
machineModules = builtins.readDir (self + "/machines");

25
pkgs/clan-cli/tests/test_import_sops_cli.py
View File

@@ -1,5 +1,7 @@
from pathlib import Path
from typing import TYPE_CHECKING
from fixtures_flakes import FlakeForTest
from clan_cli.debug import repro_env_break
import pytest
from cli import Cli
@@ -10,7 +12,7 @@ if TYPE_CHECKING:
def test_import_sops(
test_root: Path,
test_flake: Path,
test_flake: FlakeForTest,
capsys: pytest.CaptureFixture,
monkeypatch: pytest.MonkeyPatch,
age_keys: list["KeyPair"],
@@ -18,16 +20,15 @@ def test_import_sops(
cli = Cli()
monkeypatch.setenv("SOPS_AGE_KEY", age_keys[1].privkey)
cli.run(["secrets", "machines", "add", "machine1", age_keys[0].pubkey])
cli.run(["secrets", "users", "add", "user1", age_keys[1].pubkey])
cli.run(["secrets", "users", "add", "user2", age_keys[2].pubkey])
cli.run(["secrets", "groups", "add-user", "group1", "user1"])
cli.run(["secrets", "groups", "add-user", "group1", "user2"])
cli.run(["secrets", "machines", "add", "machine1", age_keys[0].pubkey, test_flake.name])
cli.run(["secrets", "users", "add", "user1", age_keys[1].pubkey, test_flake.name])
cli.run(["secrets", "users", "add", "user2", age_keys[2].pubkey, test_flake.name])
cli.run(["secrets", "groups", "add-user", "group1", "user1", test_flake.name])
cli.run(["secrets", "groups", "add-user", "group1", "user2", test_flake.name])
# To edit:
# SOPS_AGE_KEY=AGE-SECRET-KEY-1U5ENXZQAY62NC78Y2WC0SEGRRMAEEKH79EYY5TH4GPFWJKEAY0USZ6X7YQ sops --age age14tva0txcrl0zes05x7gkx56qd6wd9q3nwecjac74xxzz4l47r44sv3fz62 ./data/secrets.yaml
cli.run(
[
cmd = [
"secrets",
"import-sops",
"--group",
@@ -35,13 +36,17 @@ def test_import_sops(
"--machine",
"machine1",
str(test_root.joinpath("data", "secrets.yaml")),
test_flake.name
]
repro_env_break(work_dir=test_flake.path, cmd=cmd)
cli.run(
cmd
)
capsys.readouterr()
cli.run(["secrets", "users", "list"])
cli.run(["secrets", "users", "list", test_flake.name])
users = sorted(capsys.readouterr().out.rstrip().split())
assert users == ["user1", "user2"]
capsys.readouterr()
cli.run(["secrets", "get", "secret-key"])
cli.run(["secrets", "get", "secret-key", test_flake.name])
assert capsys.readouterr().out == "secret-value"

34
pkgs/clan-cli/tests/test_machines_api.py
View File

@@ -1,46 +1,47 @@
from pathlib import Path
from api import TestClient
from fixtures_flakes import FlakeForTest
from clan_cli.debug import repro_env_break
def test_machines(api: TestClient, test_flake: Path) -> None:
response = api.get("/api/machines")
def test_machines(api: TestClient, test_flake: FlakeForTest) -> None:
response = api.get(f"/api/{test_flake.name}/machines")
assert response.status_code == 200
assert response.json() == {"machines": []}
response = api.post("/api/machines", json={"name": "test"})
response = api.post(f"/api/{test_flake.name}/machines", json={"name": "test"})
assert response.status_code == 201
assert response.json() == {"machine": {"name": "test", "status": "unknown"}}
response = api.get("/api/machines/test")
response = api.get(f"/api/{test_flake.name}/machines/test")
assert response.status_code == 200
assert response.json() == {"machine": {"name": "test", "status": "unknown"}}
response = api.get("/api/machines")
response = api.get(f"/api/{test_flake.name}/machines")
assert response.status_code == 200
assert response.json() == {"machines": [{"name": "test", "status": "unknown"}]}
def test_configure_machine(api: TestClient, test_flake: Path) -> None:
def test_configure_machine(api: TestClient, test_flake: FlakeForTest) -> None:
# ensure error 404 if machine does not exist when accessing the config
response = api.get("/api/machines/machine1/config")
response = api.get(f"/api/{test_flake.name}/machines/machine1/config")
assert response.status_code == 404
# ensure error 404 if machine does not exist when writing to the config
response = api.put("/api/machines/machine1/config", json={})
response = api.put(f"/api/{test_flake.name}/machines/machine1/config", json={})
assert response.status_code == 404
# create the machine
response = api.post("/api/machines", json={"name": "machine1"})
response = api.post(f"/api/{test_flake.name}/machines", json={"name": "machine1"})
assert response.status_code == 201
# ensure an empty config is returned by default for a new machine
response = api.get("/api/machines/machine1/config")
response = api.get(f"/api/{test_flake.name}/machines/machine1/config")
assert response.status_code == 200
assert response.json() == {"config": {}}
# get jsonschema for machine
response = api.get("/api/machines/machine1/schema")
response = api.get(f"/api/{test_flake.name}/machines/machine1/schema")
assert response.status_code == 200
json_response = response.json()
assert "schema" in json_response and "properties" in json_response["schema"]
@@ -91,6 +92,11 @@ def test_configure_machine(api: TestClient, test_flake: Path) -> None:
devices=["/dev/fake_disk"],
),
),
f"/api/{test_flake.name}machines/machine1/config",
json=dict(
clan=dict(
jitsi=True,
)
),
)
@@ -110,8 +116,8 @@ def test_configure_machine(api: TestClient, test_flake: Path) -> None:
assert response.status_code == 200
assert response.json() == {"config": config2}
# ensure that the config has actually been updated
response = api.get("/api/machines/machine1/config")
# get the config again
response = api.get(f"/api/{test_flake.name}/machines/machine1/config")
assert response.status_code == 200
assert response.json() == {"config": config2}

6
pkgs/clan-cli/tests/test_secrets_generate.py
View File

@@ -21,8 +21,8 @@ def test_generate_secret(
monkeypatch.chdir(test_flake_with_core.path)
monkeypatch.setenv("SOPS_AGE_KEY", age_keys[0].privkey)
cli = Cli()
cli.run(["secrets", "users", "add", "user1", age_keys[0].pubkey])
cli.run(["secrets", "generate", "vm1"])
cli.run(["secrets", "users", "add", "user1", age_keys[0].pubkey, test_flake_with_core.name])
cli.run(["secrets", "generate", "vm1", test_flake_with_core.name])
has_secret(test_flake_with_core.name, "vm1-age.key")
has_secret(test_flake_with_core.name, "vm1-zerotier-identity-secret")
network_id = machine_get_fact(
@@ -43,7 +43,7 @@ def test_generate_secret(
secret1_mtime = identity_secret.lstat().st_mtime_ns
# test idempotency
cli.run(["secrets", "generate", "vm1"])
cli.run(["secrets", "generate", "vm1", test_flake_with_core.name])
assert age_key.lstat().st_mtime_ns == age_key_mtime
assert identity_secret.lstat().st_mtime_ns == secret1_mtime

16
pkgs/clan-cli/tests/test_secrets_password_store.py
View File

@@ -14,15 +14,15 @@ from clan_cli.ssh import HostGroup
def test_upload_secret(
monkeypatch: pytest.MonkeyPatch,
test_flake_with_core_and_pass: FlakeForTest,
temporary_dir: Path,
temporary_home: Path,
host_group: HostGroup,
) -> None:
monkeypatch.chdir(test_flake_with_core_and_pass.path)
gnupghome = temporary_dir / "gpg"
gnupghome = temporary_home / "gpg"
gnupghome.mkdir(mode=0o700)
monkeypatch.setenv("GNUPGHOME", str(gnupghome))
monkeypatch.setenv("PASSWORD_STORE_DIR", str(temporary_dir / "pass"))
gpg_key_spec = temporary_dir / "gpg_key_spec"
monkeypatch.setenv("PASSWORD_STORE_DIR", str(temporary_home / "pass"))
gpg_key_spec = temporary_home / "gpg_key_spec"
gpg_key_spec.write_text(
"""
Key-Type: 1
@@ -39,18 +39,18 @@ def test_upload_secret(
check=True,
)
subprocess.run(nix_shell(["pass"], ["pass", "init", "test@local"]), check=True)
cli.run(["secrets", "generate", "vm1"])
cli.run(["secrets", "generate", "vm1", test_flake_with_core_and_pass.name])
network_id = machine_get_fact(
test_flake_with_core_and_pass.name, "vm1", "zerotier-network-id"
)
assert len(network_id) == 16
identity_secret = (
temporary_dir / "pass" / "machines" / "vm1" / "zerotier-identity-secret.gpg"
temporary_home / "pass" / "machines" / "vm1" / "zerotier-identity-secret.gpg"
)
secret1_mtime = identity_secret.lstat().st_mtime_ns
# test idempotency
cli.run(["secrets", "generate", "vm1"])
cli.run(["secrets", "generate", "vm1", test_flake_with_core_and_pass.name])
assert identity_secret.lstat().st_mtime_ns == secret1_mtime
flake = test_flake_with_core_and_pass.path.joinpath("flake.nix")
@@ -58,7 +58,7 @@ def test_upload_secret(
addr = f"{host.user}@{host.host}:{host.port}?StrictHostKeyChecking=no&UserKnownHostsFile=/dev/null&IdentityFile={host.key}"
new_text = flake.read_text().replace("__CLAN_DEPLOYMENT_ADDRESS__", addr)
flake.write_text(new_text)
cli.run(["secrets", "upload", "vm1"])
cli.run(["secrets", "upload", "vm1", test_flake_with_core_and_pass.name])
zerotier_identity_secret = (
test_flake_with_core_and_pass.path / "secrets" / "zerotier-identity-secret"
)

18
pkgs/clan-cli/tests/test_secrets_upload.py
View File

@@ -3,7 +3,7 @@ from typing import TYPE_CHECKING
import pytest
from cli import Cli
from fixtures_flakes import FlakeForTest
from clan_cli.ssh import HostGroup
if TYPE_CHECKING:
@@ -13,29 +13,29 @@ if TYPE_CHECKING:
@pytest.mark.impure
def test_secrets_upload(
monkeypatch: pytest.MonkeyPatch,
test_flake_with_core: Path,
test_flake_with_core: FlakeForTest,
host_group: HostGroup,
age_keys: list["KeyPair"],
) -> None:
monkeypatch.chdir(test_flake_with_core)
monkeypatch.chdir(test_flake_with_core.path)
monkeypatch.setenv("SOPS_AGE_KEY", age_keys[0].privkey)
cli = Cli()
cli.run(["secrets", "users", "add", "user1", age_keys[0].pubkey])
cli.run(["secrets", "users", "add", "user1", age_keys[0].pubkey, test_flake_with_core.name])
cli.run(["secrets", "machines", "add", "vm1", age_keys[1].pubkey])
cli.run(["secrets", "machines", "add", "vm1", age_keys[1].pubkey, test_flake_with_core.name])
monkeypatch.setenv("SOPS_NIX_SECRET", age_keys[0].privkey)
cli.run(["secrets", "set", "vm1-age.key"])
cli.run(["secrets", "set", "vm1-age.key", test_flake_with_core.name])
flake = test_flake_with_core.joinpath("flake.nix")
flake = test_flake_with_core.path.joinpath("flake.nix")
host = host_group.hosts[0]
addr = f"{host.user}@{host.host}:{host.port}?StrictHostKeyChecking=no&UserKnownHostsFile=/dev/null&IdentityFile={host.key}"
new_text = flake.read_text().replace("__CLAN_DEPLOYMENT_ADDRESS__", addr)
flake.write_text(new_text)
cli.run(["secrets", "upload", "vm1"])
cli.run(["secrets", "upload", "vm1", test_flake_with_core.name])
# the flake defines this path as the location where the sops key should be installed
sops_key = test_flake_with_core.joinpath("key.txt")
sops_key = test_flake_with_core.path.joinpath("key.txt")
assert sops_key.exists()
assert sops_key.read_text() == age_keys[0].privkey

6
pkgs/clan-cli/tests/test_vms_api.py
View File

@@ -2,13 +2,13 @@ from pathlib import Path
import pytest
from api import TestClient
from fixtures_flakes import FlakeForTest
@pytest.mark.impure
def test_inspect(api: TestClient, test_flake_with_core: Path) -> None:
def test_inspect(api: TestClient, test_flake_with_core: FlakeForTest) -> None:
response = api.post(
"/api/vms/inspect",
json=dict(flake_url=str(test_flake_with_core), flake_attr="vm1"),
json=dict(flake_url=str(test_flake_with_core.path), flake_attr="vm1"),
)
assert response.status_code == 200, f"Failed to inspect vm: {response.text}"

20
pkgs/clan-cli/tests/test_vms_api_create.py
View File

@@ -9,6 +9,7 @@ from fixtures_flakes import FlakeForTest, create_flake
from httpx import SyncByteStream
from root import CLAN_CORE
from clan_cli.debug import repro_env_break
from clan_cli.types import FlakeName
if TYPE_CHECKING:
@@ -17,7 +18,8 @@ if TYPE_CHECKING:
@pytest.fixture
def flake_with_vm_with_secrets(
monkeypatch: pytest.MonkeyPatch, temporary_home: Path
monkeypatch: pytest.MonkeyPatch,
temporary_home: Path
) -> Iterator[FlakeForTest]:
yield from create_flake(
monkeypatch,
@@ -42,15 +44,6 @@ def remote_flake_with_vm_without_secrets(
)
@pytest.fixture
def create_user_with_age_key(
monkeypatch: pytest.MonkeyPatch,
test_flake: FlakeForTest,
age_keys: list["KeyPair"],
) -> None:
monkeypatch.setenv("SOPS_AGE_KEY", age_keys[0].privkey)
cli = Cli()
cli.run(["secrets", "users", "add", "user1", age_keys[0].pubkey, test_flake.name])
def generic_create_vm_test(api: TestClient, flake: Path, vm: str) -> None:
@@ -97,8 +90,13 @@ def test_create_local(
api: TestClient,
monkeypatch: pytest.MonkeyPatch,
flake_with_vm_with_secrets: FlakeForTest,
create_user_with_age_key: None,
age_keys: list["KeyPair"],
) -> None:
monkeypatch.setenv("SOPS_AGE_KEY", age_keys[0].privkey)
cli = Cli()
cmd = ["secrets", "users", "add", "user1", age_keys[0].pubkey, flake_with_vm_with_secrets.name]
cli.run(cmd)
generic_create_vm_test(api, flake_with_vm_with_secrets.path, "vm_with_secrets")

14
pkgs/clan-cli/tests/test_vms_cli.py
View File

@@ -1,7 +1,7 @@
import os
from pathlib import Path
from typing import TYPE_CHECKING
from fixtures_flakes import FlakeForTest
import pytest
from cli import Cli
@@ -12,9 +12,9 @@ no_kvm = not os.path.exists("/dev/kvm")
@pytest.mark.impure
def test_inspect(test_flake_with_core: Path, capsys: pytest.CaptureFixture) -> None:
def test_inspect(test_flake_with_core: FlakeForTest, capsys: pytest.CaptureFixture) -> None:
cli = Cli()
cli.run(["vms", "inspect", "vm1"])
cli.run(["vms", "inspect", "vm1", test_flake_with_core.name])
out = capsys.readouterr() # empty the buffer
assert "Cores" in out.out
@@ -23,11 +23,11 @@ def test_inspect(test_flake_with_core: Path, capsys: pytest.CaptureFixture) -> N
@pytest.mark.impure
def test_create(
monkeypatch: pytest.MonkeyPatch,
test_flake_with_core: Path,
test_flake_with_core: FlakeForTest,
age_keys: list["KeyPair"],
) -> None:
monkeypatch.chdir(test_flake_with_core)
monkeypatch.chdir(test_flake_with_core.path)
monkeypatch.setenv("SOPS_AGE_KEY", age_keys[0].privkey)
cli = Cli()
cli.run(["secrets", "users", "add", "user1", age_keys[0].pubkey])
cli.run(["vms", "create", "vm1"])
cli.run(["secrets", "users", "add", "user1", age_keys[0].pubkey, test_flake_with_core.name])
cli.run(["vms", "create", "vm1", test_flake_with_core.name])

10
pkgs/clan-cli/tests/test_webui.py
View File

@@ -10,12 +10,12 @@ from ports import PortFunction
@pytest.mark.timeout(10)
def test_start_server(unused_tcp_port: PortFunction, temporary_dir: Path) -> None:
def test_start_server(unused_tcp_port: PortFunction, temporary_home: Path) -> None:
port = unused_tcp_port()
fifo = temporary_dir / "fifo"
fifo = temporary_home / "fifo"
os.mkfifo(fifo)
notify_script = temporary_dir / "firefox"
notify_script = temporary_home / "firefox"
bash = shutil.which("bash")
assert bash is not None
notify_script.write_text(
@@ -27,8 +27,8 @@ echo "1" > {fifo}
notify_script.chmod(0o700)
env = os.environ.copy()
print(str(temporary_dir.absolute()))
env["PATH"] = ":".join([str(temporary_dir.absolute())] + env["PATH"].split(":"))
print(str(temporary_home.absolute()))
env["PATH"] = ":".join([str(temporary_home.absolute())] + env["PATH"].split(":"))
with subprocess.Popen(
[sys.executable, "-m", "clan_cli.webui", "--port", str(port)], env=env
) as p: