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

Deleted everything webui

Browse Source
This commit is contained in:
Qubasa
2023-12-14 18:47:14 +01:00
parent 1a36ef242f
commit 7dc2c21517
172 changed files with 4 additions and 57045 deletions
Download Patch File Download Diff File Expand all files Collapse all files

39
pkgs/clan-cli/README.md
View File

@@ -1,6 +1,6 @@
# clan-cli
The clan-cli contains the command line interface as well as the graphical webui through the `clan webui` command.
The clan-cli contains the command line interface
## Hacking on the cli
@@ -17,43 +17,6 @@ After you can use the local bin wrapper to test things in the cli:
./bin/clan
```
## Hacking on the webui
By default the webui is build from a tarball available https://git.clan.lol/clan/-/packages/generic/ui/.
To start a local developement environment instead, use the `--dev` flag:
```
./bin/clan webui --dev
```
This will spawn two webserver, a python one to for the api and a nodejs one that rebuilds the ui on the fly.
## Run webui directly
Useful for vscode run and debug option
```bash
python -m clan_cli.webui --reload --no-open
```
Add this `launch.json` to your .vscode directory to have working breakpoints in your vscode editor.
```json
{
"version": "0.2.0",
"configurations": [
{
"name": "Clan Webui",
"type": "python",
"request": "launch",
"module": "clan_cli.webui",
"justMyCode": true,
"args": ["--reload", "--no-open", "--log-level", "debug"]
}
]
}
```
## Run locally single-threaded for debugging
By default tests run in parallel using pytest-parallel.

37
pkgs/clan-cli/bin/gen-openapi
View File

@@ -1,37 +0,0 @@
#!/usr/bin/env python
import argparse
import json
import sys
from pathlib import Path
from uvicorn.importer import import_from_string
def main() -> None:
parser = argparse.ArgumentParser(prog="gen-openapi")
parser.add_argument(
"app", help='App import string. Eg. "main:app"', default="main:app"
)
parser.add_argument("--app-dir", help="Directory containing the app", default=None)
parser.add_argument(
"--out", help="Output file ending in .json", default="openapi.json"
)
args = parser.parse_args()
if args.app_dir is not None:
print(f"adding {args.app_dir} to sys.path")
sys.path.insert(0, args.app_dir)
print(f"importing app from {args.app}")
app = import_from_string(args.app)
openapi = app.openapi()
version = openapi.get("openapi", "unknown version")
print(f"writing openapi spec v{version}")
out = Path(args.out)
out.parent.mkdir(parents=True, exist_ok=True)
out.write_text(json.dumps(openapi, indent=2))
if __name__ == "__main__":
main()

5
pkgs/clan-cli/clan_cli/__init__.py
View File

@@ -6,7 +6,7 @@ from pathlib import Path
from types import ModuleType
from typing import Any
from . import backups, config, flakes, history, machines, secrets, vms, webui
from . import backups, config, flakes, history, machines, secrets, vms
from .custom_logger import setup_logging
from .dirs import get_clan_flake_toplevel, is_clan_flake
from .ssh import cli as ssh_cli
@@ -105,9 +105,6 @@ def create_parser(prog: str | None = None) -> argparse.ArgumentParser:
)
machines.register_parser(parser_machine)
parser_webui = subparsers.add_parser("webui", help="start webui")
webui.register_parser(parser_webui)
parser_vms = subparsers.add_parser("vms", help="manage virtual machines")
vms.register_parser(parser_vms)

64
pkgs/clan-cli/clan_cli/webui/__init__.py
View File

@@ -1,64 +0,0 @@
import argparse
from collections.abc import Callable
from typing import NoReturn
start_server: Callable | None = None
ServerImportError: ImportError | None = None
try:
from .server import start_server
except ImportError as e:
ServerImportError = e
def fastapi_is_not_installed(_: argparse.Namespace) -> NoReturn:
assert ServerImportError is not None
print(
f"Dependencies for the webserver is not installed. The webui command has been disabled ({ServerImportError})"
)
exit(1)
def register_parser(parser: argparse.ArgumentParser) -> None:
parser.add_argument("--port", type=int, default=2979, help="Port to listen on")
parser.add_argument(
"--host", type=str, default="localhost", help="Host to listen on"
)
parser.add_argument(
"--no-open", action="store_true", help="Don't open the browser", default=False
)
parser.add_argument(
"--dev", action="store_true", help="Run in development mode", default=False
)
parser.add_argument(
"--dev-port",
type=int,
default=3000,
help="Port to listen on for the dev server",
)
parser.add_argument(
"--dev-host", type=str, default="localhost", help="Host to listen on"
)
parser.add_argument(
"--reload", action="store_true", help="Don't reload on changes", default=False
)
parser.add_argument(
"--log-level",
type=str,
default="info",
help="Log level",
choices=["critical", "error", "warning", "info", "debug", "trace"],
)
parser.add_argument(
"sub_url",
type=str,
default="/",
nargs="?",
help="Sub url to open in the browser",
)
# Set the args.func variable in args
if start_server is None:
parser.set_defaults(func=fastapi_is_not_installed)
else:
parser.set_defaults(func=start_server)

15
pkgs/clan-cli/clan_cli/webui/__main__.py
View File

@@ -1,15 +0,0 @@
import argparse
from . import register_parser
if __name__ == "__main__":
# this is use in our integration test
parser = argparse.ArgumentParser()
# call the register_parser function, which adds arguments to the parser
register_parser(parser)
args = parser.parse_args()
# call the function that is stored
# in the func attribute of args, and pass args as the argument
# look into register_parser to see how this is done
args.func(args)

10
pkgs/clan-cli/clan_cli/webui/api_errors.py
View File

@@ -1,10 +0,0 @@
import logging
from pydantic import BaseModel
log = logging.getLogger(__name__)
class MissingClanImports(BaseModel):
missing_clan_imports: list[str] = []
msg: str = "Some requested clan modules could not be found"

20
pkgs/clan-cli/clan_cli/webui/api_inputs.py
View File

@@ -1,20 +0,0 @@
import logging
from pydantic import AnyUrl, BaseModel, Extra, parse_obj_as
from ..flakes.create import DEFAULT_URL
log = logging.getLogger(__name__)
class FlakeCreateInput(BaseModel):
url: AnyUrl = parse_obj_as(AnyUrl, DEFAULT_URL)
class MachineConfig(BaseModel):
clanImports: list[str] = [] # noqa: N815
clan: dict = {}
# allow extra fields to cover the full spectrum of a nixos config
class Config:
extra = Extra.allow

68
pkgs/clan-cli/clan_cli/webui/api_outputs.py
View File

@@ -1,68 +0,0 @@
from enum import Enum
from pydantic import BaseModel, Extra, Field
from ..async_cmd import CmdOut
class Status(Enum):
ONLINE = "online"
OFFLINE = "offline"
UNKNOWN = "unknown"
class ClanModulesResponse(BaseModel):
clan_modules: list[str]
class Machine(BaseModel):
name: str
status: Status
class MachinesResponse(BaseModel):
machines: list[Machine]
class MachineResponse(BaseModel):
machine: Machine
class ConfigResponse(BaseModel):
clanImports: list[str] = [] # noqa: N815
clan: dict = {}
# allow extra fields to cover the full spectrum of a nixos config
class Config:
extra = Extra.allow
class SchemaResponse(BaseModel):
schema_: dict = Field(alias="schema")
class VerifyMachineResponse(BaseModel):
success: bool
error: str | None
class FlakeAttrResponse(BaseModel):
flake_attrs: list[str]
class FlakeAction(BaseModel):
id: str
uri: str
class FlakeListResponse(BaseModel):
flakes: list[str]
class FlakeCreateResponse(BaseModel):
cmd_out: dict[str, CmdOut]
class FlakeResponse(BaseModel):
content: str
actions: list[FlakeAction]

56
pkgs/clan-cli/clan_cli/webui/app.py
View File

@@ -1,56 +0,0 @@
import logging
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from fastapi.routing import APIRoute
from fastapi.staticfiles import StaticFiles
from .assets import asset_path
from .error_handlers import clan_error_handler
from .routers import clan_modules, flake, health, machines, root
from .settings import settings
from .tags import tags_metadata
# Logging setup
log = logging.getLogger(__name__)
def setup_app() -> FastAPI:
app = FastAPI()
if settings.env.is_development():
# Allow CORS in development mode for nextjs dev server
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
app.include_router(clan_modules.router)
app.include_router(flake.router)
app.include_router(health.router)
app.include_router(machines.router)
# Needs to be last in register. Because of wildcard route
app.include_router(root.router)
app.add_exception_handler(Exception, clan_error_handler)
app.mount("/static", StaticFiles(directory=asset_path()), name="static")
# Add tag descriptions to the OpenAPI schema
app.openapi_tags = tags_metadata
for route in app.routes:
if isinstance(route, APIRoute):
route.operation_id = route.name # in this case, 'read_items'
log.debug(f"Registered route: {route}")
for i in app.exception_handlers.items():
log.debug(f"Registered exception handler: {i}")
return app
app = setup_app()

39
pkgs/clan-cli/clan_cli/webui/assets.py
View File

@@ -1,39 +0,0 @@
import functools
import logging
from pathlib import Path
log = logging.getLogger(__name__)
def get_hash(string: str) -> str:
"""
This function takes a string like '/nix/store/kkvk20b8zh8aafdnfjp6dnf062x19732-source'
and returns the hash part 'kkvk20b8zh8aafdnfjp6dnf062x19732' after '/nix/store/' and before '-source'.
"""
# Split the string by '/' and get the last element
last_element = string.split("/")[-1]
# Split the last element by '-' and get the first element
hash_part = last_element.split("-")[0]
# Return the hash part
return hash_part
def check_divergence(path: Path) -> None:
p = path.resolve()
log.info("Absolute web asset path: %s", p)
if not p.is_dir():
raise FileNotFoundError(p)
# Get the hash part of the path
gh = get_hash(str(p))
log.debug(f"Serving webui asset with hash {gh}")
@functools.cache
def asset_path() -> Path:
path = Path(__file__).parent / "assets"
log.debug("Serving assets from: %s", path)
check_divergence(path)
return path

54
pkgs/clan-cli/clan_cli/webui/error_handlers.py
View File

@@ -1,54 +0,0 @@
import logging
from fastapi import Request, status
from fastapi.encoders import jsonable_encoder
from fastapi.responses import JSONResponse
from ..errors import ClanError, ClanHttpError
from .settings import settings
log = logging.getLogger(__name__)
def clan_error_handler(request: Request, exc: Exception) -> JSONResponse:
headers = {}
if settings.env.is_development():
headers["Access-Control-Allow-Origin"] = "*"
headers["Access-Control-Allow-Methods"] = "*"
if isinstance(exc, ClanHttpError):
return JSONResponse(
status_code=exc.status_code,
content=jsonable_encoder(dict(detail={"msg": exc.msg})),
headers=headers,
)
elif isinstance(exc, ClanError):
log.error(f"ClanError: {exc}")
detail = [
{
"loc": [],
"msg": str(exc),
}
]
return JSONResponse(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
content=jsonable_encoder(dict(detail=detail)),
headers=headers,
)
else:
log.exception(f"Unhandled Exception: {exc}")
return JSONResponse(
status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
content=jsonable_encoder(
dict(
detail=[
{
"loc": [],
"msg": str(exc),
}
]
)
),
headers=headers,
)

952
pkgs/clan-cli/clan_cli/webui/openapi.json
View File

@@ -1,952 +0,0 @@
{
"openapi": "3.1.0",
"info": {
"title": "FastAPI",
"version": "0.1.0"
},
"paths": {
"/api/clan_modules": {
"get": {
"tags": ["modules"],
"summary": "List Clan Modules",
"operationId": "list_clan_modules",
"parameters": [
{
"name": "flake_dir",
"in": "query",
"required": true,
"schema": {
"title": "Flake Dir",
"type": "string",
"format": "path"
}
}
],
"responses": {
"200": {
"description": "Successful Response",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/ClanModulesResponse"
}
}
}
},
"422": {
"description": "Validation Error",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/HTTPValidationError"
}
}
}
}
}
}
},
"/api/flake/history": {
"post": {
"tags": ["flake"],
"summary": "Flake History Append",
"operationId": "flake_history_append",
"parameters": [
{
"name": "flake_dir",
"in": "query",
"required": true,
"schema": {
"title": "Flake Dir",
"type": "string",
"format": "path"
}
}
],
"responses": {
"200": {
"description": "Successful Response",
"content": {
"application/json": {
"schema": {}
}
}
},
"422": {
"description": "Validation Error",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/HTTPValidationError"
}
}
}
}
}
},
"get": {
"tags": ["flake"],
"summary": "Flake History List",
"operationId": "flake_history_list",
"responses": {
"200": {
"description": "Successful Response",
"content": {
"application/json": {
"schema": {
"title": "Response Flake History List Api Flake History Get",
"type": "array",
"items": {
"type": "string",
"format": "path"
}
}
}
}
}
}
}
},
"/api/flake/attrs": {
"get": {
"tags": ["flake"],
"summary": "Inspect Flake Attrs",
"operationId": "inspect_flake_attrs",
"parameters": [
{
"name": "url",
"in": "query",
"required": true,
"schema": {
"title": "Url",
"anyOf": [
{
"type": "string",
"minLength": 1,
"maxLength": 65536,
"format": "uri"
},
{
"type": "string",
"format": "path"
}
]
}
}
],
"responses": {
"200": {
"description": "Successful Response",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/FlakeAttrResponse"
}
}
}
},
"422": {
"description": "Validation Error",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/HTTPValidationError"
}
}
}
}
}
}
},
"/api/flake/inspect": {
"get": {
"tags": ["flake"],
"summary": "Inspect Flake",
"operationId": "inspect_flake",
"parameters": [
{
"name": "url",
"in": "query",
"required": true,
"schema": {
"title": "Url",
"anyOf": [
{
"type": "string",
"minLength": 1,
"maxLength": 65536,
"format": "uri"
},
{
"type": "string",
"format": "path"
}
]
}
}
],
"responses": {
"200": {
"description": "Successful Response",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/FlakeResponse"
}
}
}
},
"422": {
"description": "Validation Error",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/HTTPValidationError"
}
}
}
}
}
}
},
"/api/flake/create": {
"post": {
"tags": ["flake"],
"summary": "Create Flake",
"operationId": "create_flake",
"parameters": [
{
"name": "flake_dir",
"in": "query",
"required": true,
"schema": {
"title": "Flake Dir",
"type": "string",
"format": "path"
}
}
],
"requestBody": {
"required": true,
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/FlakeCreateInput"
}
}
}
},
"responses": {
"201": {
"description": "Successful Response",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/FlakeCreateResponse"
}
}
}
},
"422": {
"description": "Validation Error",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/HTTPValidationError"
}
}
}
}
}
}
},
"/api/machines": {
"get": {
"tags": ["machine"],
"summary": "List Machines",
"operationId": "list_machines",
"parameters": [
{
"name": "flake_dir",
"in": "query",
"required": true,
"schema": {
"title": "Flake Dir",
"type": "string",
"format": "path"
}
}
],
"responses": {
"200": {
"description": "Successful Response",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/MachinesResponse"
}
}
}
},
"422": {
"description": "Validation Error",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/HTTPValidationError"
}
}
}
}
}
}
},
"/api/machines/{name}": {
"get": {
"tags": ["machine"],
"summary": "Get Machine",
"operationId": "get_machine",
"parameters": [
{
"name": "name",
"in": "path",
"required": true,
"schema": {
"title": "Name",
"type": "string"
}
},
{
"name": "flake_dir",
"in": "query",
"required": true,
"schema": {
"title": "Flake Dir",
"type": "string",
"format": "path"
}
}
],
"responses": {
"200": {
"description": "Successful Response",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/MachineResponse"
}
}
}
},
"422": {
"description": "Validation Error",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/HTTPValidationError"
}
}
}
}
}
}
},
"/api/machines/{name}/config": {
"get": {
"tags": ["machine"],
"summary": "Get Machine Config",
"operationId": "get_machine_config",
"parameters": [
{
"name": "name",
"in": "path",
"required": true,
"schema": {
"title": "Name",
"type": "string"
}
},
{
"name": "flake_dir",
"in": "query",
"required": true,
"schema": {
"title": "Flake Dir",
"type": "string",
"format": "path"
}
}
],
"responses": {
"200": {
"description": "Successful Response",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/ConfigResponse"
}
}
}
},
"422": {
"description": "Validation Error",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/HTTPValidationError"
}
}
}
}
}
},
"put": {
"tags": ["machine"],
"summary": "Set Machine Config",
"operationId": "set_machine_config",
"parameters": [
{
"name": "name",
"in": "path",
"required": true,
"schema": {
"title": "Name",
"type": "string"
}
},
{
"name": "flake_dir",
"in": "query",
"required": true,
"schema": {
"title": "Flake Dir",
"type": "string",
"format": "path"
}
}
],
"requestBody": {
"required": true,
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/MachineConfig"
}
}
}
},
"responses": {
"200": {
"description": "Successful Response",
"content": {
"application/json": {
"schema": {}
}
}
},
"422": {
"description": "Validation Error",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/HTTPValidationError"
}
}
}
}
}
}
},
"/api/schema": {
"put": {
"tags": ["machine"],
"summary": "Get Machine Schema",
"operationId": "get_machine_schema",
"parameters": [
{
"name": "flake_dir",
"in": "query",
"required": true,
"schema": {
"title": "Flake Dir",
"type": "string",
"format": "path"
}
}
],
"requestBody": {
"required": true,
"content": {
"application/json": {
"schema": {
"title": "Config",
"type": "object"
}
}
}
},
"responses": {
"200": {
"description": "Successful Response",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/SchemaResponse"
}
}
}
},
"400": {
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/MissingClanImports"
}
}
},
"description": "Bad Request"
},
"422": {
"description": "Validation Error",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/HTTPValidationError"
}
}
}
}
}
}
},
"/api/machines/{name}/verify": {
"get": {
"tags": ["machine"],
"summary": "Get Verify Machine Config",
"operationId": "get_verify_machine_config",
"parameters": [
{
"name": "name",
"in": "path",
"required": true,
"schema": {
"title": "Name",
"type": "string"
}
},
{
"name": "flake_dir",
"in": "query",
"required": true,
"schema": {
"title": "Flake Dir",
"type": "string",
"format": "path"
}
}
],
"responses": {
"200": {
"description": "Successful Response",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/VerifyMachineResponse"
}
}
}
},
"422": {
"description": "Validation Error",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/HTTPValidationError"
}
}
}
}
}
},
"put": {
"tags": ["machine"],
"summary": "Put Verify Machine Config",
"operationId": "put_verify_machine_config",
"parameters": [
{
"name": "name",
"in": "path",
"required": true,
"schema": {
"title": "Name",
"type": "string"
}
},
{
"name": "flake_dir",
"in": "query",
"required": true,
"schema": {
"title": "Flake Dir",
"type": "string",
"format": "path"
}
}
],
"requestBody": {
"required": true,
"content": {
"application/json": {
"schema": {
"title": "Config",
"type": "object"
}
}
}
},
"responses": {
"200": {
"description": "Successful Response",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/VerifyMachineResponse"
}
}
}
},
"422": {
"description": "Validation Error",
"content": {
"application/json": {
"schema": {
"$ref": "#/components/schemas/HTTPValidationError"
}
}
}
}
}
}
}
},
"components": {
"schemas": {
"ClanModulesResponse": {
"properties": {
"clan_modules": {
"items": {
"type": "string"
},
"type": "array",
"title": "Clan Modules"
}
},
"type": "object",
"required": ["clan_modules"],
"title": "ClanModulesResponse"
},
"CmdOut": {
"properties": {
"stdout": {
"type": "string",
"title": "Stdout"
},
"stderr": {
"type": "string",
"title": "Stderr"
},
"cwd": {
"type": "string",
"format": "path",
"title": "Cwd"
}
},
"type": "object",
"required": ["stdout", "stderr", "cwd"],
"title": "CmdOut"
},
"ConfigResponse": {
"properties": {
"clanImports": {
"items": {
"type": "string"
},
"type": "array",
"title": "Clanimports",
"default": []
},
"clan": {
"type": "object",
"title": "Clan",
"default": {}
}
},
"type": "object",
"title": "ConfigResponse"
},
"FlakeAction": {
"properties": {
"id": {
"type": "string",
"title": "Id"
},
"uri": {
"type": "string",
"title": "Uri"
}
},
"type": "object",
"required": ["id", "uri"],
"title": "FlakeAction"
},
"FlakeAttrResponse": {
"properties": {
"flake_attrs": {
"items": {
"type": "string"
},
"type": "array",
"title": "Flake Attrs"
}
},
"type": "object",
"required": ["flake_attrs"],
"title": "FlakeAttrResponse"
},
"FlakeCreateInput": {
"properties": {
"url": {
"type": "string",
"maxLength": 65536,
"minLength": 1,
"format": "uri",
"title": "Url",
"default": "git+https://git.clan.lol/clan/clan-core?new-clan"
}
},
"type": "object",
"title": "FlakeCreateInput"
},
"FlakeCreateResponse": {
"properties": {
"cmd_out": {
"additionalProperties": {
"items": [
{
"type": "string",
"title": "Stdout"
},
{
"type": "string",
"title": "Stderr"
},
{
"type": "string",
"format": "path",
"title": "Cwd"
}
],
"type": "array",
"maxItems": 3,
"minItems": 3
},
"type": "object",
"title": "Cmd Out"
}
},
"type": "object",
"required": ["cmd_out"],
"title": "FlakeCreateResponse"
},
"FlakeResponse": {
"properties": {
"content": {
"type": "string",
"title": "Content"
},
"actions": {
"items": {
"$ref": "#/components/schemas/FlakeAction"
},
"type": "array",
"title": "Actions"
}
},
"type": "object",
"required": ["content", "actions"],
"title": "FlakeResponse"
},
"HTTPValidationError": {
"properties": {
"detail": {
"items": {
"$ref": "#/components/schemas/ValidationError"
},
"type": "array",
"title": "Detail"
}
},
"type": "object",
"title": "HTTPValidationError"
},
"Machine": {
"properties": {
"name": {
"type": "string",
"title": "Name"
},
"status": {
"$ref": "#/components/schemas/Status"
}
},
"type": "object",
"required": ["name", "status"],
"title": "Machine"
},
"MachineConfig": {
"properties": {
"clanImports": {
"items": {
"type": "string"
},
"type": "array",
"title": "Clanimports",
"default": []
},
"clan": {
"type": "object",
"title": "Clan",
"default": {}
}
},
"type": "object",
"title": "MachineConfig"
},
"MachineResponse": {
"properties": {
"machine": {
"$ref": "#/components/schemas/Machine"
}
},
"type": "object",
"required": ["machine"],
"title": "MachineResponse"
},
"MachinesResponse": {
"properties": {
"machines": {
"items": {
"$ref": "#/components/schemas/Machine"
},
"type": "array",
"title": "Machines"
}
},
"type": "object",
"required": ["machines"],
"title": "MachinesResponse"
},
"MissingClanImports": {
"properties": {
"missing_clan_imports": {
"items": {
"type": "string"
},
"type": "array",
"title": "Missing Clan Imports",
"default": []
},
"msg": {
"type": "string",
"title": "Msg",
"default": "Some requested clan modules could not be found"
}
},
"type": "object",
"title": "MissingClanImports"
},
"SchemaResponse": {
"properties": {
"schema": {
"type": "object",
"title": "Schema"
}
},
"type": "object",
"required": ["schema"],
"title": "SchemaResponse"
},
"Status": {
"enum": ["online", "offline", "unknown"],
"title": "Status",
"description": "An enumeration."
},
"ValidationError": {
"properties": {
"loc": {
"items": {
"anyOf": [
{
"type": "string"
},
{
"type": "integer"
}
]
},
"type": "array",
"title": "Location"
},
"msg": {
"type": "string",
"title": "Message"
},
"type": {
"type": "string",
"title": "Error Type"
}
},
"type": "object",
"required": ["loc", "msg", "type"],
"title": "ValidationError"
},
"VerifyMachineResponse": {
"properties": {
"success": {
"type": "boolean",
"title": "Success"
},
"error": {
"type": "string",
"title": "Error"
}
},
"type": "object",
"required": ["success"],
"title": "VerifyMachineResponse"
}
}
},
"tags": [
{
"name": "flake",
"description": "Operations on a flake.",
"externalDocs": {
"description": "What is a flake?",
"url": "https://www.tweag.io/blog/2020-05-25-flakes/"
}
},
{
"name": "machine",
"description": "Manage physical machines. Instances of a flake"
},
{
"name": "vm",
"description": "Manage virtual machines. Instances of a flake"
},
{
"name": "modules",
"description": "Manage cLAN modules of a flake"
}
]
}

0
pkgs/clan-cli/clan_cli/webui/routers/__init__.py
View File

23
pkgs/clan-cli/clan_cli/webui/routers/clan_modules.py
View File

@@ -1,23 +0,0 @@
# Logging setup
import logging
from pathlib import Path
from fastapi import APIRouter, HTTPException
from clan_cli.clan_modules import get_clan_module_names
from ..api_outputs import (
ClanModulesResponse,
)
from ..tags import Tags
log = logging.getLogger(__name__)
router = APIRouter()
@router.get("/api/clan_modules", tags=[Tags.modules])
async def list_clan_modules(flake_dir: Path) -> ClanModulesResponse:
module_names, error = get_clan_module_names(flake_dir)
if error is not None:
raise HTTPException(status_code=400, detail=error)
return ClanModulesResponse(clan_modules=module_names)

93
pkgs/clan-cli/clan_cli/webui/routers/flake.py
View File

@@ -1,93 +0,0 @@
import json
from json.decoder import JSONDecodeError
from pathlib import Path
from typing import Annotated
from fastapi import APIRouter, Body, HTTPException, status
from pydantic import AnyUrl
from clan_cli.webui.api_inputs import (
FlakeCreateInput,
)
from clan_cli.webui.api_outputs import (
FlakeAction,
FlakeAttrResponse,
FlakeCreateResponse,
FlakeResponse,
)
from ...async_cmd import run
from ...flakes import create
from ...nix import nix_command, nix_flake_show
from ..tags import Tags
router = APIRouter()
# TODO: Check for directory traversal
async def get_attrs(url: AnyUrl | Path) -> list[str]:
cmd = nix_flake_show(url)
out = await run(cmd)
data: dict[str, dict] = {}
try:
data = json.loads(out.stdout)
except JSONDecodeError:
raise HTTPException(status_code=422, detail="Could not load flake.")
nixos_configs = data.get("nixosConfigurations", {})
flake_attrs = list(nixos_configs.keys())
if not flake_attrs:
raise HTTPException(
status_code=422, detail="No entry or no attribute: nixosConfigurations"
)
return flake_attrs
# TODO: Check for directory traversal
@router.get("/api/flake/attrs", tags=[Tags.flake])
async def inspect_flake_attrs(url: AnyUrl | Path) -> FlakeAttrResponse:
return FlakeAttrResponse(flake_attrs=await get_attrs(url))
# TODO: Check for directory traversal
@router.get("/api/flake/inspect", tags=[Tags.flake])
async def inspect_flake(
url: AnyUrl | Path,
) -> FlakeResponse:
actions = []
# Extract the flake from the given URL
# We do this by running 'nix flake prefetch {url} --json'
cmd = nix_command(["flake", "prefetch", str(url), "--json", "--refresh"])
out = await run(cmd)
data: dict[str, str] = json.loads(out.stdout)
if data.get("storePath") is None:
raise HTTPException(status_code=500, detail="Could not load flake")
content: str
with open(Path(data.get("storePath", "")) / Path("flake.nix")) as f:
content = f.read()
# TODO: Figure out some measure when it is insecure to inspect or create a VM
actions.append(FlakeAction(id="vms/inspect", uri="api/vms/inspect"))
actions.append(FlakeAction(id="vms/create", uri="api/vms/create"))
return FlakeResponse(content=content, actions=actions)
@router.post(
"/api/flake/create", tags=[Tags.flake], status_code=status.HTTP_201_CREATED
)
async def create_flake(
flake_dir: Path, args: Annotated[FlakeCreateInput, Body()]
) -> FlakeCreateResponse:
if flake_dir.exists():
raise HTTPException(
status_code=status.HTTP_409_CONFLICT,
detail="Flake already exists",
)
cmd_out = await create.create_flake(flake_dir, args.url)
return FlakeCreateResponse(cmd_out=cmd_out)

8
pkgs/clan-cli/clan_cli/webui/routers/health.py
View File

@@ -1,8 +0,0 @@
from fastapi import APIRouter
router = APIRouter()
@router.get("/health", include_in_schema=False)
async def health() -> str:
return "OK"

92
pkgs/clan-cli/clan_cli/webui/routers/machines.py
View File

@@ -1,92 +0,0 @@
# Logging setup
import logging
from pathlib import Path
from typing import Annotated
from fastapi import APIRouter, Body
from fastapi.encoders import jsonable_encoder
from clan_cli.webui.api_errors import MissingClanImports
from clan_cli.webui.api_inputs import MachineConfig
from ...config.machine import (
config_for_machine,
set_config_for_machine,
verify_machine_config,
)
from ...config.schema import machine_schema
from ...machines.list import list_machines as _list_machines
from ..api_outputs import (
ConfigResponse,
Machine,
MachineResponse,
MachinesResponse,
SchemaResponse,
Status,
VerifyMachineResponse,
)
from ..tags import Tags
log = logging.getLogger(__name__)
router = APIRouter()
@router.get("/api/machines", tags=[Tags.machine])
async def list_machines(flake_dir: Path) -> MachinesResponse:
machines = []
for m in _list_machines(flake_dir):
machines.append(Machine(name=m, status=Status.UNKNOWN))
return MachinesResponse(machines=machines)
@router.get("/api/machines/{name}", tags=[Tags.machine])
async def get_machine(flake_dir: Path, name: str) -> MachineResponse:
log.error("TODO")
return MachineResponse(machine=Machine(name=name, status=Status.UNKNOWN))
@router.get("/api/machines/{name}/config", tags=[Tags.machine])
async def get_machine_config(flake_dir: Path, name: str) -> ConfigResponse:
config = config_for_machine(flake_dir, name)
return ConfigResponse(**config)
@router.put("/api/machines/{name}/config", tags=[Tags.machine])
async def set_machine_config(
flake_dir: Path, name: str, config: Annotated[MachineConfig, Body()]
) -> None:
conf = jsonable_encoder(config)
set_config_for_machine(flake_dir, name, conf)
@router.put(
"/api/schema",
tags=[Tags.machine],
responses={400: {"model": MissingClanImports}},
)
async def get_machine_schema(
flake_dir: Path, config: Annotated[dict, Body()]
) -> SchemaResponse:
schema = machine_schema(flake_dir, config=config)
return SchemaResponse(schema=schema)
@router.get("/api/machines/{name}/verify", tags=[Tags.machine])
async def get_verify_machine_config(
flake_dir: Path, name: str
) -> VerifyMachineResponse:
error = verify_machine_config(flake_dir, name)
success = error is None
return VerifyMachineResponse(success=success, error=error)
@router.put("/api/machines/{name}/verify", tags=[Tags.machine])
async def put_verify_machine_config(
flake_dir: Path,
name: str,
config: Annotated[dict, Body()],
) -> VerifyMachineResponse:
error = verify_machine_config(flake_dir, name, config)
success = error is None
return VerifyMachineResponse(success=success, error=error)

36
pkgs/clan-cli/clan_cli/webui/routers/root.py
View File

@@ -1,36 +0,0 @@
import logging
import os
from mimetypes import guess_type
from pathlib import Path
from fastapi import APIRouter, Response
from ..assets import asset_path
router = APIRouter()
log = logging.getLogger(__name__)
@router.get("/{path_name:path}", include_in_schema=False)
async def root(path_name: str) -> Response:
if path_name == "":
path_name = "index.html"
filename = Path(os.path.normpath(asset_path() / path_name))
if not filename.is_relative_to(asset_path()):
log.error("Prevented directory traversal: %s", filename)
# prevent directory traversal
return Response(status_code=403)
if not filename.is_file():
if filename.suffix == "":
filename = filename.with_suffix(".html")
if not filename.is_file():
log.error("File not found: %s", filename)
return Response(status_code=404)
else:
return Response(status_code=404)
content_type, _ = guess_type(filename)
return Response(filename.read_bytes(), media_type=content_type)

105
pkgs/clan-cli/clan_cli/webui/server.py
View File

@@ -1,105 +0,0 @@
import argparse
import logging
import os
import shutil
import subprocess
import time
import urllib.request
from collections.abc import Iterator
from contextlib import ExitStack, contextmanager
from pathlib import Path
from threading import Thread
# XXX: can we dynamically load this using nix develop?
import uvicorn
from pydantic import AnyUrl, IPvAnyAddress
from pydantic.tools import parse_obj_as
from clan_cli.errors import ClanError
log = logging.getLogger(__name__)
def open_browser(base_url: AnyUrl, sub_url: str) -> None:
for i in range(5):
try:
urllib.request.urlopen(base_url + "/health")
break
except OSError:
time.sleep(i)
url = parse_obj_as(AnyUrl, f"{base_url}/{sub_url.removeprefix('/')}")
_open_browser(url)
def _open_browser(url: AnyUrl) -> subprocess.Popen:
for browser in ("firefox", "iceweasel", "iceape", "seamonkey"):
if shutil.which(browser):
# Do not add a new profile, as it will break in combination with
# the -kiosk flag.
cmd = [
browser,
"-kiosk",
"-new-window",
url,
]
print(" ".join(cmd))
return subprocess.Popen(cmd)
for browser in ("chromium", "chromium-browser", "google-chrome", "chrome"):
if shutil.which(browser):
return subprocess.Popen([browser, f"--app={url}"])
raise ClanError("No browser found")
@contextmanager
def spawn_node_dev_server(host: IPvAnyAddress, port: int) -> Iterator[None]:
log.info("Starting node dev server...")
path = Path(__file__).parent.parent.parent.parent / "ui"
with subprocess.Popen(
[
"direnv",
"exec",
path,
"npm",
"run",
"dev",
"--",
"--hostname",
str(host),
"--port",
str(port),
],
cwd=path,
) as proc:
try:
yield
finally:
proc.terminate()
def start_server(args: argparse.Namespace) -> None:
os.environ["CLAN_WEBUI_ENV"] = "development" if args.dev else "production"
with ExitStack() as stack:
headers: list[tuple[str, str]] = []
if args.dev:
stack.enter_context(spawn_node_dev_server(args.dev_host, args.dev_port))
base_url = f"http://{args.dev_host}:{args.dev_port}"
host = args.dev_host
if ":" in host:
host = f"[{host}]"
else:
base_url = f"http://{args.host}:{args.port}"
if not args.no_open:
Thread(target=open_browser, args=(base_url, args.sub_url)).start()
uvicorn.run(
"clan_cli.webui.app:app",
host=args.host,
port=args.port,
log_level=args.log_level,
reload=args.reload,
access_log=args.log_level == "debug",
headers=headers,
)

32
pkgs/clan-cli/clan_cli/webui/settings.py
View File

@@ -1,32 +0,0 @@
import logging
import os
from enum import Enum
log = logging.getLogger(__name__)
class EnvType(Enum):
production = "production"
development = "development"
@staticmethod
def from_environment() -> "EnvType":
t = os.environ.get("CLAN_WEBUI_ENV", "production")
try:
return EnvType[t]
except KeyError:
log.warning(f"Invalid environment type: {t}, fallback to production")
return EnvType.production
def is_production(self) -> bool:
return self == EnvType.production
def is_development(self) -> bool:
return self == EnvType.development
class Settings:
env = EnvType.from_environment()
settings = Settings()

37
pkgs/clan-cli/clan_cli/webui/tags.py
View File

@@ -1,37 +0,0 @@
from enum import Enum
from typing import Any
class Tags(Enum):
flake = "flake"
machine = "machine"
vm = "vm"
modules = "modules"
root = "root"
def __str__(self) -> str:
return self.value
tags_metadata: list[dict[str, Any]] = [
{
"name": str(Tags.flake),
"description": "Operations on a flake.",
"externalDocs": {
"description": "What is a flake?",
"url": "https://www.tweag.io/blog/2020-05-25-flakes/",
},
},
{
"name": str(Tags.machine),
"description": "Manage physical machines. Instances of a flake",
},
{
"name": str(Tags.vm),
"description": "Manage virtual machines. Instances of a flake",
},
{
"name": str(Tags.modules),
"description": "Manage cLAN modules of a flake",
},
]

27
pkgs/clan-cli/default.nix
View File

@@ -1,8 +1,6 @@
{ age
, lib
, argcomplete
, fastapi
, uvicorn
, installShellFiles
, nix
, openssh
@@ -21,7 +19,6 @@
, wheel
, fakeroot
, rsync
, ui-assets
, bash
, sshpass
, zbar
@@ -36,7 +33,6 @@
, rope
, clan-core-path
, writeShellScriptBin
, nodePackages
}:
let
@@ -45,8 +41,6 @@ let
];
pytestDependencies = runtimeDependencies ++ dependencies ++ [
fastapi # optional dependencies: if not enabled, webui subcommand will not work
uvicorn # optional dependencies: if not enabled, webui subcommand will not work
#schemathesis # optional for http fuzzing
pytest
@@ -93,7 +87,6 @@ let
rm $out/clan_cli/config/jsonschema
ln -s ${nixpkgs'} $out/clan_cli/nixpkgs
cp -r ${../../lib/jsonschema} $out/clan_cli/config/jsonschema
ln -s ${ui-assets} $out/clan_cli/webui/assets
'';
nixpkgs' = runCommand "nixpkgs" { nativeBuildInputs = [ nix ]; } ''
mkdir $out
@@ -168,28 +161,8 @@ python3.pkgs.buildPythonApplication {
fi
touch $out
'';
check-clan-openapi = runCommand "check-clan-openapi" { } ''
export PATH=${checkPython}/bin:$PATH
${checkPython}/bin/python ${source}/bin/gen-openapi --out ./openapi.json --app-dir ${source} clan_cli.webui.app:app
${lib.getExe nodePackages.prettier} --write ./openapi.json
if ! diff -u ./openapi.json ${source}/clan_cli/webui/openapi.json; then
echo "nix run .#update-clan-openapi to update the openapi.json file."
exit 1
fi
touch $out
'';
};
passthru.update-clan-openapi = writeShellScriptBin "update-clan-openapi" ''
export PATH=${checkPython}/bin:$PATH
git_root=$(git rev-parse --show-toplevel)
cd "$git_root/pkgs/clan-cli"
${checkPython}/bin/python ./bin/gen-openapi --out clan_cli/webui/openapi.json --app-dir . clan_cli.webui.app:app
${lib.getExe nodePackages.prettier} --write clan_cli/webui/openapi.json
'';
passthru.nixpkgs = nixpkgs';
passthru.checkPython = checkPython;

8
pkgs/clan-cli/flake-module.nix
View File

@@ -30,12 +30,11 @@
in
{
devShells.clan-cli = pkgs.callPackage ./shell.nix {
inherit (self'.packages) clan-cli ui-assets nix-unit;
inherit (self'.packages) clan-cli nix-unit;
# inherit (inputs) democlan;
};
packages = {
clan-cli = pkgs.python3.pkgs.callPackage ./default.nix {
inherit (self'.packages) ui-assets;
inherit (inputs) nixpkgs;
# inherit (inputs) democlan;
inherit (inputs.nixpkgs-for-deal.legacyPackages.${system}.python3Packages) deal;
@@ -45,11 +44,6 @@
default = self'.packages.clan-cli;
};
apps.update-clan-openapi = {
type = "app";
program = "${self'.packages.clan-cli.passthru.update-clan-openapi}/bin/update-clan-openapi";
};
checks = self'.packages.clan-cli.tests;
};

3
pkgs/clan-cli/shell.nix
View File

@@ -1,4 +1,4 @@
{ nix-unit, clan-cli, ui-assets, system, mkShell, writeScriptBin, openssh, ruff, python3 }:
{ nix-unit, clan-cli, system, mkShell, writeScriptBin, openssh, ruff, python3 }:
let
checkScript = writeScriptBin "check" ''
nix build .#checks.${system}.{treefmt,clan-pytest} -L "$@"
@@ -39,7 +39,6 @@ mkShell {
--editable $repo_root
ln -sfT ${clan-cli.nixpkgs} clan_cli/nixpkgs
ln -sfT ${ui-assets} clan_cli/webui/assets
export PATH="$tmp_path/python/bin:${checkScript}/bin:$PATH"
export PYTHONPATH="$repo_root:$tmp_path/python/${pythonWithDeps.sitePackages}:"

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

@@ -1,14 +0,0 @@
import logging
import pytest
from fastapi.testclient import TestClient
from clan_cli.webui.app import app
# TODO: Why stateful
@pytest.fixture(scope="session")
def api() -> TestClient:
# logging.getLogger("httpx").setLevel(level=logging.WARNING)
logging.getLogger("asyncio").setLevel(logging.INFO)
return TestClient(app, raise_server_exceptions=False)

1
pkgs/clan-cli/tests/conftest.py
View File

@@ -10,7 +10,6 @@ from clan_cli.nix import nix_shell
sys.path.append(os.path.join(os.path.dirname(__file__), "helpers"))
pytest_plugins = [
"api",
"temporary_dir",
"root",
"age_keys",

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

@@ -9,8 +9,6 @@ from pathlib import Path
from typing import NamedTuple
import pytest
from pydantic import AnyUrl
from pydantic.tools import parse_obj_as
from root import CLAN_CORE
from clan_cli.dirs import nixpkgs_source
@@ -136,16 +134,6 @@ def test_local_democlan(
yield FlakeForTest(democlan_p)
@pytest.fixture
def test_democlan_url(
monkeypatch: pytest.MonkeyPatch, temporary_home: Path
) -> Iterator[AnyUrl]:
yield parse_obj_as(
AnyUrl,
"https://git.clan.lol/clan/democlan/archive/main.tar.gz",
)
@pytest.fixture
def test_flake_with_core_and_pass(
monkeypatch: pytest.MonkeyPatch, temporary_home: Path

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

@@ -1,16 +0,0 @@
import pytest
from api import TestClient
from fixtures_flakes import FlakeForTest
@pytest.mark.with_core
def test_configure_machine(api: TestClient, test_flake_with_core: FlakeForTest) -> None:
# retrieve the list of available clanModules
response = api.get(f"/api/clan_modules?flake_dir={test_flake_with_core.path}")
assert response.status_code == 200, response.text
response_json = response.json()
assert isinstance(response_json, dict)
assert "clan_modules" in response_json
assert len(response_json["clan_modules"]) > 0
# ensure all entries are a string
assert all(isinstance(x, str) for x in response_json["clan_modules"])

21
pkgs/clan-cli/tests/test_create_flake.py
View File

@@ -3,35 +3,14 @@ import subprocess
from pathlib import Path
import pytest
from api import TestClient
from cli import Cli
from clan_cli.flakes.create import DEFAULT_URL
@pytest.fixture
def cli() -> Cli:
return Cli()
@pytest.mark.impure
def test_create_flake_api(
monkeypatch: pytest.MonkeyPatch, api: TestClient, temporary_home: Path
) -> None:
flake_dir = temporary_home / "test-flake"
response = api.post(
f"/api/flake/create?flake_dir={flake_dir}",
json=dict(
flake_dir=str(flake_dir),
url=str(DEFAULT_URL),
),
)
assert response.status_code == 201, f"Failed to create flake {response.text}"
assert (flake_dir / ".clan-flake").exists()
assert (flake_dir / "flake.nix").exists()
@pytest.mark.impure
def test_create_flake(
monkeypatch: pytest.MonkeyPatch,

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

@@ -1,54 +0,0 @@
import json
import logging
import pytest
from api import TestClient
from fixtures_flakes import FlakeForTest
log = logging.getLogger(__name__)
@pytest.mark.impure
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,
)
assert response.status_code == 200, "Failed to inspect vm"
data = response.json()
print("Data: ", data)
assert data.get("flake_attrs") == ["vm1", "vm2"]
@pytest.mark.impure
def test_inspect_err(api: TestClient) -> None:
params = {"url": "flake-parts"}
response = api.get(
"/api/flake/attrs",
params=params,
)
assert response.status_code != 200, "Succeed to inspect vm but expected to fail"
data = response.json()
print("Data: ", data)
assert data.get("detail")
@pytest.mark.impure
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/inspect",
params=params,
)
assert response.status_code == 200, "Failed to inspect vm"
data = response.json()
print("Data: ", json.dumps(data, indent=2))
assert data.get("content") is not None
actions = data.get("actions")
assert actions is not None
assert len(actions) == 2
assert actions[0].get("id") == "vms/inspect"
assert actions[0].get("uri") == "api/vms/inspect"
assert actions[1].get("id") == "vms/create"
assert actions[1].get("uri") == "api/vms/create"

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

@@ -1,10 +0,0 @@
import pytest
from api import TestClient
@pytest.mark.impure
def test_static_files(api: TestClient) -> None:
response = api.get("/")
assert response.headers["content-type"] == "text/html; charset=utf-8"
response = api.get("/does-no-exists.txt")
assert response.status_code == 404

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

@@ -1,64 +0,0 @@
import os
import select
import shutil
import subprocess
import sys
from pathlib import Path
import pytest
from cli import Cli
from ports import PortFunction
@pytest.mark.timeout(10)
def test_start_server(unused_tcp_port: PortFunction, temporary_home: Path) -> None:
Cli()
port = unused_tcp_port()
fifo = temporary_home / "fifo"
os.mkfifo(fifo)
# Create a script called "firefox" in the temporary home directory that
# writes "1" to the fifo. This is used to notify the test that the firefox has been
# started.
notify_script = temporary_home / "firefox"
bash = shutil.which("bash")
assert bash is not None
notify_script.write_text(
f"""#!{bash}
set -x
echo "1" > {fifo}
"""
)
notify_script.chmod(0o700)
# Add the temporary home directory to the PATH so that the script is found
env = os.environ.copy()
env["PATH"] = f"{temporary_home}:{env['PATH']}"
# Add build/src to PYTHONPATH so that the webui module is found in nix sandbox
# TODO: We need a way to make sure things which work in the devshell also work in the sandbox
python_path = env.get("PYTHONPATH")
if python_path:
env["PYTHONPATH"] = f"/build/src:{python_path}"
# breakpoint_container(
# cmd=[sys.executable, "-m", "clan_cli.webui", "--port", str(port)],
# env=env,
# work_dir=temporary_home,
# )
with subprocess.Popen(
[sys.executable, "-m", "clan_cli.webui", "--port", str(port)],
env=env,
stdout=sys.stderr,
stderr=sys.stderr,
text=True,
) as p:
try:
with open(fifo) as f:
r, _, _ = select.select([f], [], [], 10)
assert f in r
assert f.read().strip() == "1"
finally:
p.kill()