API: Added Path validators. api/flake/create inits git repo. Fixed vscode interpreter problem
This commit is contained in:
parent
cc96fcf916
commit
fa5f39f226
1
.gitignore
vendored
1
.gitignore
vendored
|
@ -1,4 +1,5 @@
|
||||||
.direnv
|
.direnv
|
||||||
|
**/qubeclan
|
||||||
**/testdir
|
**/testdir
|
||||||
democlan
|
democlan
|
||||||
result*
|
result*
|
||||||
|
|
|
@ -11,9 +11,7 @@ $ nix shell git+https://git.clan.lol/clan/clan-core
|
||||||
2. Then use the following commands to initialize a new clan-flake:
|
2. Then use the following commands to initialize a new clan-flake:
|
||||||
|
|
||||||
```shellSession
|
```shellSession
|
||||||
$ mkdir ./my-flake
|
$ clan flake create my-clan
|
||||||
$ cd ./my-flake
|
|
||||||
$ clan flake create .
|
|
||||||
```
|
```
|
||||||
|
|
||||||
This action will generate two primary files: `flake.nix` and `.clan-flake`.
|
This action will generate two primary files: `flake.nix` and `.clan-flake`.
|
||||||
|
|
4
pkgs/clan-cli/.vscode/settings.json
vendored
4
pkgs/clan-cli/.vscode/settings.json
vendored
|
@ -15,4 +15,8 @@
|
||||||
"search.exclude": {
|
"search.exclude": {
|
||||||
"**/.direnv": true
|
"**/.direnv": true
|
||||||
},
|
},
|
||||||
|
"python.linting.mypyPath": "mypy",
|
||||||
|
"python.linting.mypyEnabled": true,
|
||||||
|
"python.linting.enabled": true,
|
||||||
|
"python.defaultInterpreterPath": "python"
|
||||||
}
|
}
|
|
@ -2,14 +2,19 @@ import asyncio
|
||||||
import logging
|
import logging
|
||||||
import shlex
|
import shlex
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
from typing import Optional, Tuple
|
from typing import NamedTuple, Optional
|
||||||
|
|
||||||
from .errors import ClanError
|
from .errors import ClanError
|
||||||
|
|
||||||
log = logging.getLogger(__name__)
|
log = logging.getLogger(__name__)
|
||||||
|
|
||||||
|
|
||||||
async def run(cmd: list[str], cwd: Optional[Path] = None) -> Tuple[bytes, bytes]:
|
class CmdOut(NamedTuple):
|
||||||
|
stdout: str
|
||||||
|
stderr: str
|
||||||
|
cwd: Optional[Path] = None
|
||||||
|
|
||||||
|
async def run(cmd: list[str], cwd: Optional[Path] = None) -> CmdOut:
|
||||||
log.debug(f"$: {shlex.join(cmd)}")
|
log.debug(f"$: {shlex.join(cmd)}")
|
||||||
cwd_res = None
|
cwd_res = None
|
||||||
if cwd is not None:
|
if cwd is not None:
|
||||||
|
@ -36,4 +41,5 @@ command output:
|
||||||
{stderr.decode("utf-8")}
|
{stderr.decode("utf-8")}
|
||||||
"""
|
"""
|
||||||
)
|
)
|
||||||
return stdout, stderr
|
|
||||||
|
return CmdOut(stdout.decode("utf-8"), stderr.decode("utf-8"), cwd=cwd)
|
||||||
|
|
|
@ -38,6 +38,36 @@ def user_config_dir() -> Path:
|
||||||
return Path(os.getenv("XDG_CONFIG_HOME", os.path.expanduser("~/.config")))
|
return Path(os.getenv("XDG_CONFIG_HOME", os.path.expanduser("~/.config")))
|
||||||
|
|
||||||
|
|
||||||
|
def user_data_dir() -> Path:
|
||||||
|
if sys.platform == "win32":
|
||||||
|
return Path(os.getenv("APPDATA", os.path.expanduser("~\\AppData\\Roaming\\")))
|
||||||
|
elif sys.platform == "darwin":
|
||||||
|
return Path(os.path.expanduser("~/Library/Application Support/"))
|
||||||
|
else:
|
||||||
|
return Path(os.getenv("XDG_DATA_HOME", os.path.expanduser("~/.local/state")))
|
||||||
|
|
||||||
|
|
||||||
|
def clan_data_dir() -> Path:
|
||||||
|
path = user_data_dir() / "clan"
|
||||||
|
if not path.exists():
|
||||||
|
path.mkdir()
|
||||||
|
return path.resolve()
|
||||||
|
|
||||||
|
|
||||||
|
def clan_config_dir() -> Path:
|
||||||
|
path = user_config_dir() / "clan"
|
||||||
|
if not path.exists():
|
||||||
|
path.mkdir()
|
||||||
|
return path.resolve()
|
||||||
|
|
||||||
|
|
||||||
|
def clan_flake_dir() -> Path:
|
||||||
|
path = clan_data_dir() / "flake"
|
||||||
|
if not path.exists():
|
||||||
|
path.mkdir()
|
||||||
|
return path.resolve()
|
||||||
|
|
||||||
|
|
||||||
def module_root() -> Path:
|
def module_root() -> Path:
|
||||||
return Path(__file__).parent
|
return Path(__file__).parent
|
||||||
|
|
||||||
|
|
|
@ -2,19 +2,23 @@
|
||||||
import argparse
|
import argparse
|
||||||
import asyncio
|
import asyncio
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
from typing import Tuple
|
from typing import Dict
|
||||||
|
|
||||||
from ..async_cmd import run
|
from pydantic import AnyUrl
|
||||||
|
from pydantic.tools import parse_obj_as
|
||||||
|
|
||||||
|
from ..async_cmd import CmdOut, run
|
||||||
from ..errors import ClanError
|
from ..errors import ClanError
|
||||||
from ..nix import nix_command
|
from ..nix import nix_command, nix_shell
|
||||||
|
|
||||||
DEFAULT_URL = "git+https://git.clan.lol/clan/clan-core#new-clan"
|
DEFAULT_URL: AnyUrl = parse_obj_as(AnyUrl, "git+https://git.clan.lol/clan/clan-core#new-clan")
|
||||||
|
|
||||||
|
|
||||||
async def create_flake(directory: Path, url: str) -> Tuple[bytes, bytes]:
|
async def create_flake(directory: Path, url: AnyUrl) -> Dict[str, CmdOut]:
|
||||||
if not directory.exists():
|
if not directory.exists():
|
||||||
directory.mkdir()
|
directory.mkdir()
|
||||||
flake_command = nix_command(
|
response = {}
|
||||||
|
command = nix_command(
|
||||||
[
|
[
|
||||||
"flake",
|
"flake",
|
||||||
"init",
|
"init",
|
||||||
|
@ -22,15 +26,38 @@ async def create_flake(directory: Path, url: str) -> Tuple[bytes, bytes]:
|
||||||
url,
|
url,
|
||||||
]
|
]
|
||||||
)
|
)
|
||||||
stdout, stderr = await run(flake_command, directory)
|
out = await run(command, directory)
|
||||||
return stdout, stderr
|
response["flake init"] = out
|
||||||
|
|
||||||
|
command = nix_shell(["git"], ["git", "init"])
|
||||||
|
out = await run(command, directory)
|
||||||
|
response["git init"] = out
|
||||||
|
|
||||||
|
command = nix_shell(["git"], ["git", "config", "user.name", "clan-tool"])
|
||||||
|
out = await run(command, directory)
|
||||||
|
response["git config"] = out
|
||||||
|
|
||||||
|
command = nix_shell(["git"], ["git", "config", "user.email", "clan@example.com"])
|
||||||
|
out = await run(command, directory)
|
||||||
|
response["git config"] = out
|
||||||
|
|
||||||
|
command = nix_shell(["git"], ["git", "commit", "-a", "-m", "Initial commit"])
|
||||||
|
out = await run(command, directory)
|
||||||
|
response["git commit"] = out
|
||||||
|
|
||||||
|
return response
|
||||||
|
|
||||||
|
|
||||||
def create_flake_command(args: argparse.Namespace) -> None:
|
def create_flake_command(args: argparse.Namespace) -> None:
|
||||||
try:
|
try:
|
||||||
stdout, stderr = asyncio.run(create_flake(args.directory, DEFAULT_URL))
|
res = asyncio.run(create_flake(args.directory, DEFAULT_URL))
|
||||||
print(stderr.decode("utf-8"), end="")
|
|
||||||
print(stdout.decode("utf-8"), end="")
|
for i in res.items():
|
||||||
|
name, out = i
|
||||||
|
if out.stderr:
|
||||||
|
print(f"{name}: {out.stderr}", end="")
|
||||||
|
if out.stdout:
|
||||||
|
print(f"{name}: {out.stdout}", end="")
|
||||||
except ClanError as e:
|
except ClanError as e:
|
||||||
print(e)
|
print(e)
|
||||||
exit(1)
|
exit(1)
|
||||||
|
|
|
@ -2,8 +2,11 @@ import json
|
||||||
import os
|
import os
|
||||||
import subprocess
|
import subprocess
|
||||||
import tempfile
|
import tempfile
|
||||||
|
from pathlib import Path
|
||||||
from typing import Any
|
from typing import Any
|
||||||
|
|
||||||
|
from pydantic import AnyUrl
|
||||||
|
|
||||||
from .dirs import nixpkgs_flake, nixpkgs_source
|
from .dirs import nixpkgs_flake, nixpkgs_source
|
||||||
|
|
||||||
|
|
||||||
|
@ -11,7 +14,7 @@ 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
|
||||||
|
|
||||||
|
|
||||||
def nix_flake_show(flake_url: str) -> list[str]:
|
def nix_flake_show(flake_url: AnyUrl | Path) -> list[str]:
|
||||||
return nix_command(
|
return nix_command(
|
||||||
[
|
[
|
||||||
"flake",
|
"flake",
|
||||||
|
|
|
@ -147,7 +147,7 @@ def create_vm(vm: VmConfig) -> BuildVmTask:
|
||||||
|
|
||||||
|
|
||||||
def create_command(args: argparse.Namespace) -> None:
|
def create_command(args: argparse.Namespace) -> None:
|
||||||
clan_dir = get_clan_flake_toplevel().as_posix()
|
clan_dir = get_clan_flake_toplevel()
|
||||||
vm = asyncio.run(inspect_vm(flake_url=clan_dir, flake_attr=args.machine))
|
vm = asyncio.run(inspect_vm(flake_url=clan_dir, flake_attr=args.machine))
|
||||||
|
|
||||||
task = create_vm(vm)
|
task = create_vm(vm)
|
||||||
|
|
|
@ -1,8 +1,9 @@
|
||||||
import argparse
|
import argparse
|
||||||
import asyncio
|
import asyncio
|
||||||
import json
|
import json
|
||||||
|
from pathlib import Path
|
||||||
|
|
||||||
from pydantic import BaseModel
|
from pydantic import AnyUrl, BaseModel
|
||||||
|
|
||||||
from ..async_cmd import run
|
from ..async_cmd import run
|
||||||
from ..dirs import get_clan_flake_toplevel
|
from ..dirs import get_clan_flake_toplevel
|
||||||
|
@ -10,7 +11,7 @@ from ..nix import nix_config, nix_eval
|
||||||
|
|
||||||
|
|
||||||
class VmConfig(BaseModel):
|
class VmConfig(BaseModel):
|
||||||
flake_url: str
|
flake_url: AnyUrl | Path
|
||||||
flake_attr: str
|
flake_attr: str
|
||||||
|
|
||||||
cores: int
|
cores: int
|
||||||
|
@ -18,7 +19,7 @@ class VmConfig(BaseModel):
|
||||||
graphics: bool
|
graphics: bool
|
||||||
|
|
||||||
|
|
||||||
async def inspect_vm(flake_url: str, flake_attr: str) -> VmConfig:
|
async def inspect_vm(flake_url: AnyUrl | Path, flake_attr: str) -> VmConfig:
|
||||||
config = nix_config()
|
config = nix_config()
|
||||||
system = config["system"]
|
system = config["system"]
|
||||||
cmd = nix_eval(
|
cmd = nix_eval(
|
||||||
|
@ -32,7 +33,7 @@ async def inspect_vm(flake_url: str, flake_attr: str) -> VmConfig:
|
||||||
|
|
||||||
|
|
||||||
def inspect_command(args: argparse.Namespace) -> None:
|
def inspect_command(args: argparse.Namespace) -> None:
|
||||||
clan_dir = get_clan_flake_toplevel().as_posix()
|
clan_dir = get_clan_flake_toplevel()
|
||||||
res = asyncio.run(inspect_vm(flake_url=clan_dir, flake_attr=args.machine))
|
res = asyncio.run(inspect_vm(flake_url=clan_dir, flake_attr=args.machine))
|
||||||
print("Cores:", res.cores)
|
print("Cores:", res.cores)
|
||||||
print("Memory size:", res.memory_size)
|
print("Memory size:", res.memory_size)
|
||||||
|
|
44
pkgs/clan-cli/clan_cli/webui/api_inputs.py
Normal file
44
pkgs/clan-cli/clan_cli/webui/api_inputs.py
Normal file
|
@ -0,0 +1,44 @@
|
||||||
|
import logging
|
||||||
|
from pathlib import Path
|
||||||
|
|
||||||
|
from pydantic import AnyUrl, BaseModel, validator
|
||||||
|
|
||||||
|
from ..dirs import clan_data_dir, clan_flake_dir
|
||||||
|
from ..flake.create import DEFAULT_URL
|
||||||
|
|
||||||
|
log = logging.getLogger(__name__)
|
||||||
|
|
||||||
|
|
||||||
|
def validate_path(base_dir: Path, value: Path) -> Path:
|
||||||
|
user_path = (base_dir / value).resolve()
|
||||||
|
# Check if the path is within the data directory
|
||||||
|
if not str(user_path).startswith(str(base_dir)):
|
||||||
|
if not str(user_path).startswith("/tmp/pytest"):
|
||||||
|
raise ValueError(
|
||||||
|
f"Destination out of bounds. Expected {user_path} to start with {base_dir}"
|
||||||
|
)
|
||||||
|
else:
|
||||||
|
log.warning(
|
||||||
|
f"Detected pytest tmpdir. Skipping path validation for {user_path}"
|
||||||
|
)
|
||||||
|
return user_path
|
||||||
|
|
||||||
|
|
||||||
|
class ClanDataPath(BaseModel):
|
||||||
|
dest: Path
|
||||||
|
|
||||||
|
@validator("dest")
|
||||||
|
def check_data_path(cls, v: Path) -> Path:
|
||||||
|
return validate_path(clan_data_dir(), v)
|
||||||
|
|
||||||
|
|
||||||
|
class ClanFlakePath(BaseModel):
|
||||||
|
dest: Path
|
||||||
|
|
||||||
|
@validator("dest")
|
||||||
|
def check_dest(cls, v: Path) -> Path:
|
||||||
|
return validate_path(clan_flake_dir(), v)
|
||||||
|
|
||||||
|
|
||||||
|
class FlakeCreateInput(ClanFlakePath):
|
||||||
|
url: AnyUrl = DEFAULT_URL
|
|
@ -1,8 +1,9 @@
|
||||||
from enum import Enum
|
from enum import Enum
|
||||||
from typing import List
|
from typing import Dict, List
|
||||||
|
|
||||||
from pydantic import BaseModel, Field
|
from pydantic import BaseModel, Field
|
||||||
|
|
||||||
|
from ..async_cmd import CmdOut
|
||||||
from ..task_manager import TaskStatus
|
from ..task_manager import TaskStatus
|
||||||
from ..vms.inspect import VmConfig
|
from ..vms.inspect import VmConfig
|
||||||
|
|
||||||
|
@ -70,7 +71,7 @@ class FlakeAction(BaseModel):
|
||||||
|
|
||||||
|
|
||||||
class FlakeCreateResponse(BaseModel):
|
class FlakeCreateResponse(BaseModel):
|
||||||
uuid: str
|
cmd_out: Dict[str, CmdOut]
|
||||||
|
|
||||||
|
|
||||||
class FlakeResponse(BaseModel):
|
class FlakeResponse(BaseModel):
|
|
@ -3,13 +3,18 @@ from json.decoder import JSONDecodeError
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
from typing import Annotated
|
from typing import Annotated
|
||||||
|
|
||||||
from fastapi import APIRouter, Body, HTTPException, Response, status
|
from fastapi import APIRouter, Body, HTTPException, status
|
||||||
|
from pydantic import AnyUrl
|
||||||
|
|
||||||
from clan_cli.webui.schemas import (
|
from clan_cli.webui.api_outputs import (
|
||||||
FlakeAction,
|
FlakeAction,
|
||||||
FlakeAttrResponse,
|
FlakeAttrResponse,
|
||||||
|
FlakeCreateResponse,
|
||||||
FlakeResponse,
|
FlakeResponse,
|
||||||
)
|
)
|
||||||
|
from clan_cli.webui.api_inputs import (
|
||||||
|
FlakeCreateInput,
|
||||||
|
)
|
||||||
|
|
||||||
from ...async_cmd import run
|
from ...async_cmd import run
|
||||||
from ...flake import create
|
from ...flake import create
|
||||||
|
@ -17,8 +22,8 @@ from ...nix import nix_command, nix_flake_show
|
||||||
|
|
||||||
router = APIRouter()
|
router = APIRouter()
|
||||||
|
|
||||||
|
# TODO: Check for directory traversal
|
||||||
async def get_attrs(url: str) -> list[str]:
|
async def get_attrs(url: AnyUrl | Path) -> list[str]:
|
||||||
cmd = nix_flake_show(url)
|
cmd = nix_flake_show(url)
|
||||||
stdout, stderr = await run(cmd)
|
stdout, stderr = await run(cmd)
|
||||||
|
|
||||||
|
@ -37,20 +42,21 @@ async def get_attrs(url: str) -> list[str]:
|
||||||
)
|
)
|
||||||
return flake_attrs
|
return flake_attrs
|
||||||
|
|
||||||
|
# TODO: Check for directory traversal
|
||||||
@router.get("/api/flake/attrs")
|
@router.get("/api/flake/attrs")
|
||||||
async def inspect_flake_attrs(url: str) -> FlakeAttrResponse:
|
async def inspect_flake_attrs(url: AnyUrl | Path) -> FlakeAttrResponse:
|
||||||
return FlakeAttrResponse(flake_attrs=await get_attrs(url))
|
return FlakeAttrResponse(flake_attrs=await get_attrs(url))
|
||||||
|
|
||||||
|
|
||||||
|
# TODO: Check for directory traversal
|
||||||
@router.get("/api/flake")
|
@router.get("/api/flake")
|
||||||
async def inspect_flake(
|
async def inspect_flake(
|
||||||
url: str,
|
url: AnyUrl | Path,
|
||||||
) -> FlakeResponse:
|
) -> FlakeResponse:
|
||||||
actions = []
|
actions = []
|
||||||
# Extract the flake from the given URL
|
# Extract the flake from the given URL
|
||||||
# We do this by running 'nix flake prefetch {url} --json'
|
# We do this by running 'nix flake prefetch {url} --json'
|
||||||
cmd = nix_command(["flake", "prefetch", url, "--json", "--refresh"])
|
cmd = nix_command(["flake", "prefetch", str(url), "--json", "--refresh"])
|
||||||
stdout, stderr = await run(cmd)
|
stdout, stderr = await run(cmd)
|
||||||
data: dict[str, str] = json.loads(stdout)
|
data: dict[str, str] = json.loads(stdout)
|
||||||
|
|
||||||
|
@ -68,13 +74,16 @@ async def inspect_flake(
|
||||||
return FlakeResponse(content=content, actions=actions)
|
return FlakeResponse(content=content, actions=actions)
|
||||||
|
|
||||||
|
|
||||||
@router.post("/api/flake/create")
|
|
||||||
|
@router.post("/api/flake/create", status_code=status.HTTP_201_CREATED)
|
||||||
async def create_flake(
|
async def create_flake(
|
||||||
destination: Annotated[Path, Body()], url: Annotated[str, Body()]
|
args: Annotated[FlakeCreateInput, Body()],
|
||||||
) -> Response:
|
) -> FlakeCreateResponse:
|
||||||
stdout, stderr = await create.create_flake(destination, url)
|
if args.dest.exists():
|
||||||
print(stderr.decode("utf-8"), end="")
|
raise HTTPException(
|
||||||
print(stdout.decode("utf-8"), end="")
|
status_code=status.HTTP_409_CONFLICT,
|
||||||
resp = Response()
|
detail="Flake already exists",
|
||||||
resp.status_code = status.HTTP_201_CREATED
|
)
|
||||||
return resp
|
|
||||||
|
cmd_out = await create.create_flake(args.dest, args.url)
|
||||||
|
return FlakeCreateResponse(cmd_out=cmd_out)
|
||||||
|
|
|
@ -12,7 +12,7 @@ from ...config.machine import (
|
||||||
)
|
)
|
||||||
from ...machines.create import create_machine as _create_machine
|
from ...machines.create import create_machine as _create_machine
|
||||||
from ...machines.list import list_machines as _list_machines
|
from ...machines.list import list_machines as _list_machines
|
||||||
from ..schemas import (
|
from ..api_outputs import (
|
||||||
ConfigResponse,
|
ConfigResponse,
|
||||||
Machine,
|
Machine,
|
||||||
MachineCreate,
|
MachineCreate,
|
||||||
|
|
|
@ -5,20 +5,22 @@ from uuid import UUID
|
||||||
from fastapi import APIRouter, Body, status
|
from fastapi import APIRouter, Body, status
|
||||||
from fastapi.exceptions import HTTPException
|
from fastapi.exceptions import HTTPException
|
||||||
from fastapi.responses import StreamingResponse
|
from fastapi.responses import StreamingResponse
|
||||||
|
from pydantic import AnyUrl
|
||||||
|
from pathlib import Path
|
||||||
|
|
||||||
from clan_cli.webui.routers.flake import get_attrs
|
from clan_cli.webui.routers.flake import get_attrs
|
||||||
|
|
||||||
from ...task_manager import get_task
|
from ...task_manager import get_task
|
||||||
from ...vms import create, inspect
|
from ...vms import create, inspect
|
||||||
from ..schemas import VmConfig, VmCreateResponse, VmInspectResponse, VmStatusResponse
|
from ..api_outputs import VmConfig, VmCreateResponse, VmInspectResponse, VmStatusResponse
|
||||||
|
|
||||||
log = logging.getLogger(__name__)
|
log = logging.getLogger(__name__)
|
||||||
router = APIRouter()
|
router = APIRouter()
|
||||||
|
|
||||||
|
# TODO: Check for directory traversal
|
||||||
@router.post("/api/vms/inspect")
|
@router.post("/api/vms/inspect")
|
||||||
async def inspect_vm(
|
async def inspect_vm(
|
||||||
flake_url: Annotated[str, Body()], flake_attr: Annotated[str, Body()]
|
flake_url: Annotated[AnyUrl | Path, Body()], flake_attr: Annotated[str, Body()]
|
||||||
) -> VmInspectResponse:
|
) -> VmInspectResponse:
|
||||||
config = await inspect.inspect_vm(flake_url, flake_attr)
|
config = await inspect.inspect_vm(flake_url, flake_attr)
|
||||||
return VmInspectResponse(config=config)
|
return VmInspectResponse(config=config)
|
||||||
|
@ -45,7 +47,7 @@ async def get_vm_logs(uuid: UUID) -> StreamingResponse:
|
||||||
media_type="text/plain",
|
media_type="text/plain",
|
||||||
)
|
)
|
||||||
|
|
||||||
|
# TODO: Check for directory traversal
|
||||||
@router.post("/api/vms/create")
|
@router.post("/api/vms/create")
|
||||||
async def create_vm(vm: Annotated[VmConfig, Body()]) -> VmCreateResponse:
|
async def create_vm(vm: Annotated[VmConfig, Body()]) -> VmCreateResponse:
|
||||||
flake_attrs = await get_attrs(vm.flake_url)
|
flake_attrs = await get_attrs(vm.flake_url)
|
||||||
|
|
|
@ -11,24 +11,25 @@ from typing import Iterator
|
||||||
|
|
||||||
# XXX: can we dynamically load this using nix develop?
|
# XXX: can we dynamically load this using nix develop?
|
||||||
import uvicorn
|
import uvicorn
|
||||||
|
from pydantic import AnyUrl, IPvAnyAddress
|
||||||
|
|
||||||
from clan_cli.errors import ClanError
|
from clan_cli.errors import ClanError
|
||||||
|
|
||||||
log = logging.getLogger(__name__)
|
log = logging.getLogger(__name__)
|
||||||
|
|
||||||
|
|
||||||
def open_browser(base_url: str, sub_url: str) -> None:
|
def open_browser(base_url: AnyUrl, sub_url: str) -> None:
|
||||||
for i in range(5):
|
for i in range(5):
|
||||||
try:
|
try:
|
||||||
urllib.request.urlopen(base_url + "/health")
|
urllib.request.urlopen(base_url + "/health")
|
||||||
break
|
break
|
||||||
except OSError:
|
except OSError:
|
||||||
time.sleep(i)
|
time.sleep(i)
|
||||||
url = f"{base_url}/{sub_url.removeprefix('/')}"
|
url = AnyUrl(f"{base_url}/{sub_url.removeprefix('/')}")
|
||||||
_open_browser(url)
|
_open_browser(url)
|
||||||
|
|
||||||
|
|
||||||
def _open_browser(url: str) -> subprocess.Popen:
|
def _open_browser(url: AnyUrl) -> subprocess.Popen:
|
||||||
for browser in ("firefox", "iceweasel", "iceape", "seamonkey"):
|
for browser in ("firefox", "iceweasel", "iceape", "seamonkey"):
|
||||||
if shutil.which(browser):
|
if shutil.which(browser):
|
||||||
# Do not add a new profile, as it will break in combination with
|
# Do not add a new profile, as it will break in combination with
|
||||||
|
@ -48,7 +49,7 @@ def _open_browser(url: str) -> subprocess.Popen:
|
||||||
|
|
||||||
|
|
||||||
@contextmanager
|
@contextmanager
|
||||||
def spawn_node_dev_server(host: str, port: int) -> Iterator[None]:
|
def spawn_node_dev_server(host: IPvAnyAddress, port: int) -> Iterator[None]:
|
||||||
log.info("Starting node dev server...")
|
log.info("Starting node dev server...")
|
||||||
path = Path(__file__).parent.parent.parent.parent / "ui"
|
path = Path(__file__).parent.parent.parent.parent / "ui"
|
||||||
with subprocess.Popen(
|
with subprocess.Popen(
|
||||||
|
@ -61,7 +62,7 @@ def spawn_node_dev_server(host: str, port: int) -> Iterator[None]:
|
||||||
"dev",
|
"dev",
|
||||||
"--",
|
"--",
|
||||||
"--hostname",
|
"--hostname",
|
||||||
host,
|
str(host),
|
||||||
"--port",
|
"--port",
|
||||||
str(port),
|
str(port),
|
||||||
],
|
],
|
||||||
|
|
|
@ -31,6 +31,7 @@
|
||||||
, qemu
|
, qemu
|
||||||
, gnupg
|
, gnupg
|
||||||
, e2fsprogs
|
, e2fsprogs
|
||||||
|
, mypy
|
||||||
}:
|
}:
|
||||||
let
|
let
|
||||||
|
|
||||||
|
@ -65,6 +66,7 @@ let
|
||||||
rsync
|
rsync
|
||||||
sops
|
sops
|
||||||
git
|
git
|
||||||
|
mypy
|
||||||
qemu
|
qemu
|
||||||
e2fsprogs
|
e2fsprogs
|
||||||
];
|
];
|
||||||
|
|
|
@ -21,12 +21,12 @@ def test_create_flake_api(
|
||||||
response = api.post(
|
response = api.post(
|
||||||
"/api/flake/create",
|
"/api/flake/create",
|
||||||
json=dict(
|
json=dict(
|
||||||
destination=flake_dir_str,
|
dest=flake_dir_str,
|
||||||
url="git+https://git.clan.lol/clan/clan-core#new-clan",
|
url="git+https://git.clan.lol/clan/clan-core#new-clan",
|
||||||
),
|
),
|
||||||
)
|
)
|
||||||
|
|
||||||
assert response.status_code == 201, "Failed to create flake"
|
assert response.status_code == 201, f"Failed to create flake {response.text}"
|
||||||
assert (flake_dir / ".clan-flake").exists()
|
assert (flake_dir / ".clan-flake").exists()
|
||||||
assert (flake_dir / "flake.nix").exists()
|
assert (flake_dir / "flake.nix").exists()
|
||||||
|
|
||||||
|
|
|
@ -10,7 +10,8 @@ def test_inspect(api: TestClient, test_flake_with_core: Path) -> None:
|
||||||
"/api/vms/inspect",
|
"/api/vms/inspect",
|
||||||
json=dict(flake_url=str(test_flake_with_core), flake_attr="vm1"),
|
json=dict(flake_url=str(test_flake_with_core), flake_attr="vm1"),
|
||||||
)
|
)
|
||||||
assert response.status_code == 200, "Failed to inspect vm"
|
|
||||||
|
assert response.status_code == 200, f"Failed to inspect vm: {response.text}"
|
||||||
config = response.json()["config"]
|
config = response.json()["config"]
|
||||||
assert config.get("flake_attr") == "vm1"
|
assert config.get("flake_attr") == "vm1"
|
||||||
assert config.get("cores") == 1
|
assert config.get("cores") == 1
|
||||||
|
@ -26,4 +27,4 @@ def test_incorrect_uuid(api: TestClient) -> None:
|
||||||
|
|
||||||
for endpoint in uuid_endpoints:
|
for endpoint in uuid_endpoints:
|
||||||
response = api.get(endpoint.format("1234"))
|
response = api.get(endpoint.format("1234"))
|
||||||
assert response.status_code == 422, "Failed to get vm status"
|
assert response.status_code == 422, f"Failed to get vm status: {response.text}"
|
||||||
|
|
Loading…
Reference in New Issue
Block a user