Skip to content
Snippets Groups Projects
Verified Commit 89de0cf3 authored by Karel van Klink's avatar Karel van Klink :smiley_cat:
Browse files

add docstrings to the rest of the

parent 2d9a9d55
No related branches found
No related tags found
1 merge request!111Feature/ruff everything party hat emoji
"""Workflows for the site subscription object."""
"""A creation workflow for adding a new site to the service database."""
from orchestrator.forms import FormPage
from orchestrator.targets import Target
from orchestrator.types import FormGenerator, State, SubscriptionLifecycle, UUIDstr
......@@ -20,6 +22,7 @@ from gso.utils.helpers import (
def initial_input_form_generator(product_name: str) -> FormGenerator:
"""Get input from the operator about the new site subscription."""
class CreateSiteForm(FormPage):
class Config:
title = product_name
......@@ -38,17 +41,20 @@ def initial_input_form_generator(product_name: str) -> FormGenerator:
@validator("site_ts_address", allow_reuse=True)
def validate_ts_address(cls, site_ts_address: str) -> str:
"""Validate that a terminal server address is valid."""
validate_site_fields_is_unique("site_ts_address", site_ts_address)
validate_ipv4_or_ipv6(site_ts_address)
return site_ts_address
@validator("site_country_code", allow_reuse=True)
def country_code_must_exist(cls, country_code: str) -> str:
"""Validate that the country code exists."""
validate_country_code(country_code)
return country_code
@validator("site_internal_id", "site_bgp_community_id", allow_reuse=True)
def validate_unique_fields(cls, value: str, field: ModelField) -> str | int:
"""Validate that the internal and :term:`BGP` community IDs are unique."""
return validate_site_fields_is_unique(field.name, value)
@validator("site_name", allow_reuse=True)
......@@ -69,6 +75,7 @@ def initial_input_form_generator(product_name: str) -> FormGenerator:
@step("Create subscription")
def create_subscription(product: UUIDstr, customer: UUIDstr) -> State:
"""Create a new subscription object in the service database."""
subscription = site.SiteInactive.from_product_id(product, customer)
return {
......@@ -91,6 +98,7 @@ def initialize_subscription(
site_ts_address: str,
site_tier: site_pb.SiteTier,
) -> State:
"""Initialise the subscription object with all user input."""
subscription.site.site_name = site_name
subscription.site.site_city = site_city
subscription.site.site_country = site_country
......@@ -115,6 +123,7 @@ def initialize_subscription(
target=Target.CREATE,
)
def create_site() -> StepList:
"""Create a new site subscription."""
return (
init
>> create_subscription
......
"""A modification workflow for a site."""
from orchestrator.forms import FormPage
from orchestrator.targets import Target
from orchestrator.types import FormGenerator, State, SubscriptionLifecycle, UUIDstr
......@@ -20,6 +22,7 @@ from gso.utils.helpers import validate_ipv4_or_ipv6, validate_site_fields_is_uni
def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
"""Gather input from the operator on what to change about the selected site subscription."""
subscription = Site.from_subscription(subscription_id)
class ModifySiteForm(FormPage):
......@@ -65,6 +68,7 @@ def modify_site_subscription(
site_internal_id: int,
site_ts_address: str,
) -> State:
"""Update the subscription model in the service database."""
subscription.site.site_city = site_city
subscription.site.site_latitude = site_latitude
subscription.site.site_longitude = site_longitude
......@@ -83,6 +87,10 @@ def modify_site_subscription(
target=Target.MODIFY,
)
def modify_site() -> StepList:
"""Modify a site subscription.
* Update the subscription model in the service database
"""
return (
init
>> store_process_subscription(Target.MODIFY)
......
"""A workflow for terminating a site subscription."""
from orchestrator.forms import FormPage
from orchestrator.forms.validators import Label
from orchestrator.targets import Target
......@@ -15,6 +17,7 @@ from gso.products.product_types.site import Site
def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
"""Ask the user for confirmation whether to terminate the selected site."""
Site.from_subscription(subscription_id)
class TerminateForm(FormPage):
......@@ -30,6 +33,7 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
target=Target.TERMINATE,
)
def terminate_site() -> StepList:
"""Terminate a site subscription."""
return (
init
>> store_process_subscription(Target.TERMINATE)
......
......@@ -102,6 +102,7 @@ ban-relative-imports = "all"
[tool.ruff.per-file-ignores]
"test/*" = ["ARG001", "D", "S101", "PLR2004"]
"setup.py" = ["D100"]
[tool.ruff.isort]
known-third-party = ["pydantic", "migrations"]
......
......@@ -7,10 +7,10 @@ class MockedNetboxClient:
def get_device_by_name(self):
return self.BaseMockObject(id=1, name="test")
def get_available_lags(self) -> list[str]:
def get_available_lags(self) -> list[str]: # noqa: PLR6301
return [f"LAG{lag}" for lag in range(1, 5)]
def get_available_interfaces(self):
def get_available_interfaces(self): # noqa: PLR6301
interfaces = []
for interface in range(5):
interface_data = {
......@@ -30,14 +30,14 @@ class MockedNetboxClient:
def reserve_interface(self):
return self.BaseMockObject(id=1, name="test")
def allocate_interface(self):
def allocate_interface(self): # noqa: PLR6301
return {"id": 1, "name": "test"}
def free_interface(self):
return self.BaseMockObject(id=1, name="test")
def detach_interfaces_from_lag(self):
def detach_interfaces_from_lag(self): # noqa: PLR6301
return None
def delete_interface(self):
def delete_interface(self): # noqa: PLR6301
return None
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment