Added more deal tests

This commit is contained in:
Qubasa
2023-11-21 16:44:38 +01:00
parent af4e991f87
commit 9698e57fa6
5 changed files with 47 additions and 12 deletions

View File

@@ -78,6 +78,9 @@ Let's get your development environment up and running:
``` ```
- Wait for the frontend to build. - Wait for the frontend to build.
NOTE: If you have the error "@clan/colors.json" you executed `npm install`, please do not do that. `direnv reload` will handle dependency management. Please delete node_modules with `rm -rf node_modules`.
9. **Start the Frontend**: 9. **Start the Frontend**:
- To start the frontend, execute: - To start the frontend, execute:
```bash ```bash

View File

@@ -1,8 +1,9 @@
import asyncio import asyncio
import logging import logging
import deal
import shlex import shlex
from pathlib import Path from pathlib import Path
from typing import Any, Callable, Coroutine, Dict, NamedTuple, Optional from typing import Any, Callable, Coroutine, Dict, NamedTuple, Optional, Awaitable
from .custom_logger import get_caller from .custom_logger import get_caller
from .errors import ClanError from .errors import ClanError
@@ -15,8 +16,8 @@ class CmdOut(NamedTuple):
stderr: str stderr: str
cwd: Optional[Path] = None cwd: Optional[Path] = None
@deal.raises(ClanError)
async def run(cmd: list[str], cwd: Optional[Path] = None) -> CmdOut: async def run(cmd: list[str], cwd: Optional[Path] = None) -> Awaitable[CmdOut]:
cwd_res = None cwd_res = None
if cwd is not None: if cwd is not None:
if not cwd.exists(): if not cwd.exists():
@@ -51,6 +52,7 @@ stdout:
return CmdOut(stdout.decode("utf-8"), stderr.decode("utf-8"), cwd=cwd) return CmdOut(stdout.decode("utf-8"), stderr.decode("utf-8"), cwd=cwd)
@deal.raises(ClanError)
def runforcli( def runforcli(
func: Callable[..., Coroutine[Any, Any, Dict[str, CmdOut]]], *args: Any func: Callable[..., Coroutine[Any, Any, Dict[str, CmdOut]]], *args: Any
) -> None: ) -> None:

View File

@@ -2,12 +2,13 @@ import json
import os import os
import subprocess import subprocess
import tempfile import tempfile
import deal
from pathlib import Path from pathlib import Path
from typing import Any from typing import Any
from .dirs import nixpkgs_flake, nixpkgs_source from .dirs import nixpkgs_flake, nixpkgs_source
@deal.raises(ClanError)
def nix_command(flags: list[str]) -> list[str]: def nix_command(flags: list[str]) -> list[str]:
return ["nix", "--extra-experimental-features", "nix-command flakes"] + flags return ["nix", "--extra-experimental-features", "nix-command flakes"] + flags
@@ -24,7 +25,7 @@ def nix_flake_show(flake_url: str | Path) -> list[str]:
] ]
) )
@deal.raises(ClanError)
def nix_build( def nix_build(
flags: list[str], flags: list[str],
) -> list[str]: ) -> list[str]:
@@ -40,7 +41,7 @@ def nix_build(
+ flags + flags
) )
@deal.raises(ClanError)
def nix_config() -> dict[str, Any]: def nix_config() -> dict[str, Any]:
cmd = nix_command(["show-config", "--json"]) cmd = nix_command(["show-config", "--json"])
proc = subprocess.run(cmd, check=True, text=True, stdout=subprocess.PIPE) proc = subprocess.run(cmd, check=True, text=True, stdout=subprocess.PIPE)
@@ -50,7 +51,7 @@ def nix_config() -> dict[str, Any]:
config[key] = value["value"] config[key] = value["value"]
return config return config
@deal.raises(ClanError)
def nix_eval(flags: list[str]) -> list[str]: def nix_eval(flags: list[str]) -> list[str]:
default_flags = nix_command( default_flags = nix_command(
[ [
@@ -77,7 +78,7 @@ def nix_eval(flags: list[str]) -> list[str]:
) )
return default_flags + flags return default_flags + flags
@deal.raises(ClanError)
def nix_shell(packages: list[str], cmd: list[str]) -> list[str]: def nix_shell(packages: list[str], cmd: list[str]) -> list[str]:
# we cannot use nix-shell inside the nix sandbox # we cannot use nix-shell inside the nix sandbox
# in our tests we just make sure we have all the packages # in our tests we just make sure we have all the packages

View File

@@ -186,10 +186,13 @@ def get_task(uuid: UUID) -> BaseTask:
T = TypeVar("T", bound="BaseTask") T = TypeVar("T", bound="BaseTask")
@deal.raises(ClanError)
def create_task(task_type: Type[T], *args: Any) -> T: def create_task(task_type: Type[T], *args: Any) -> T:
global POOL global POOL
# check if task_type is a callable
if not callable(task_type):
raise ClanError("task_type must be callable")
uuid = uuid4() uuid = uuid4()
task = task_type(uuid, *args) task = task_type(uuid, *args)

View File

@@ -1,8 +1,34 @@
import deal import deal
from clan_cli.task_manager import get_task from clan_cli import task_manager
from clan_cli import async_cmd
from clan_cli import nix
@deal.cases(task_manager.get_task)
@deal.cases(get_task)
def test_get_task(case: deal.TestCase) -> None: def test_get_task(case: deal.TestCase) -> None:
case() case()
@deal.cases(task_manager.create_task)
def test_create_task(case: deal.TestCase) -> None:
case()
@deal.cases(nix.nix_command)
def test_nix_command(case: deal.TestCase) -> None:
case()
@deal.cases(nix.nix_build)
def test_nix_build(case: deal.TestCase) -> None:
case()
@deal.cases(nix.nix_config)
def test_nix_config(case: deal.TestCase) -> None:
case()
@deal.cases(nix.nix_eval)
def test_nix_eval(case: deal.TestCase) -> None:
case()
@deal.cases(nix.nix_shell)
def test_nix_shell(case: deal.TestCase) -> None:
case()