Skip to content
Snippets Groups Projects
Commit 77b366b2 authored by Karel van Klink's avatar Karel van Klink :smiley_cat: Committed by JORGE SASIAIN
Browse files

resolve last linting errors

parent 7db4c25c
No related branches found
No related tags found
1 merge request!108NAT-315: Restore ISIS should be optional in migrate iptrunk
Showing
with 37 additions and 27 deletions
"""Initialisation class for the :term:`GSO` :term:`API`."""
from fastapi import APIRouter
from gso.api.v1 import router as router_v1
......
"""Version 1 of the :term:`GSO` :term:`API`."""
from fastapi import APIRouter
from gso.api.v1.imports import router as imports_router
......
""":term:`GSO` :term:`API` endpoints that import different types of existing services."""
import ipaddress
from typing import Any
from uuid import UUID
......
""":term:`API` endpoint for fetching different types of subscriptions."""
from typing import Any
from fastapi import Depends, status
......
""":term:`CLI` command for importing sites."""
import typer
app: typer.Typer = typer.Typer()
......
"""A :term:`CLI` for interacting with Netbox."""
import typer
from pynetbox import RequestError
......
......@@ -4,6 +4,7 @@
Whenever a new product type is added, this should be reflected in the :py:class:`gso.products.ProductType`
enumerator.
"""
from orchestrator.domain import SUBSCRIPTION_MODEL_REGISTRY
from pydantic_forms.types import strEnum
......
......@@ -36,7 +36,7 @@ class IptrunkType(strEnum):
T_co = TypeVar("T_co", covariant=True)
class LAGMemberList(UniqueConstrainedList[T_co]):
class LAGMemberList(UniqueConstrainedList[T_co]): # type: ignore[type-var]
"""A list of :term:`LAG` member interfaces."""
......@@ -66,7 +66,7 @@ class IptrunkInterfaceBlock(IptrunkInterfaceBlockProvisioning, lifecycle=[Subscr
interface_description: str
class IptrunkSides(UniqueConstrainedList[T_co]):
class IptrunkSides(UniqueConstrainedList[T_co]): # type: ignore[type-var]
"""A list of IP trunk interfaces that make up one side of a link."""
min_items = 2
......
"""Product block for :class:`Router` products."""
import ipaddress
from orchestrator.domain.base import ProductBlockModel
......
"""The product block that describes a site subscription."""
import re
from orchestrator.domain.base import ProductBlockModel
......
"""Contain all methods to communicate with the NetBox API endpoint. Data Center Infrastructure Main (DCIM)."""
from uuid import UUID
import pydantic
......
......@@ -2,6 +2,7 @@
:term:`LSO` is responsible for executing Ansible playbooks, that deploy subscriptions.
"""
import json
import logging
from functools import partial
......
......@@ -213,7 +213,7 @@ def validate_site_name(site_name: str) -> str:
if not pattern.match(site_name):
msg = (
"Enter a valid site name. It must consist of three uppercase letters (A-Z) followed by an optional single "
"digit (0-9)."
f"digit (0-9). Received: {site_name}"
)
raise ValueError(msg)
return site_name
"""Initialisation class that imports all workflows into :term:`GSO`."""
from orchestrator.workflows import LazyWorkflowInstance
LazyWorkflowInstance("gso.workflows.iptrunk.create_iptrunk", "create_iptrunk")
......
......@@ -174,7 +174,7 @@ def create_subscription(product: UUIDstr, customer: UUIDstr) -> State:
@step("Get information from IPAM")
def get_info_from_ipam(subscription: IptrunkProvisioning) -> State:
"""Allocate :term:`IP` resources in :term:`IPAM`."""
"""Allocate IP resources in :term:`IPAM`."""
subscription.iptrunk.iptrunk_ipv4_network = infoblox.allocate_v4_network(
"TRUNK",
subscription.iptrunk.iptrunk_description,
......
......@@ -302,9 +302,7 @@ def confirm_continue_move_fiber() -> FormGenerator:
class Config:
title = "Please confirm before continuing"
info_label: Label = (
"New trunk interface has been deployed, wait for the physical connection to be moved." # type: ignore[assignment]
)
info_label: Label = "New trunk interface has been deployed, wait for the physical connection to be moved." # type: ignore[assignment]
yield ProvisioningResultPage
......@@ -355,9 +353,7 @@ def confirm_continue_restore_isis() -> FormGenerator:
class Config:
title = "Please confirm before continuing"
info_label: Label = (
"ISIS config has been deployed, confirm if you want to restore the old metric." # type: ignore[assignment]
)
info_label: Label = "ISIS config has been deployed, confirm if you want to restore the old metric." # type: ignore[assignment]
yield ProvisioningResultPage
......
......@@ -39,8 +39,8 @@ def initialize_ae_members(subscription: Iptrunk, initial_user_input: dict, side_
iptrunk_speed = initial_user_input["iptrunk_speed"]
class NokiaLAGMember(LAGMember):
interface_name: (
available_interfaces_choices_including_current_members( # type: ignore[valid-type]
interface_name: ( # type: ignore[valid-type]
available_interfaces_choices_including_current_members(
router.owner_subscription_id,
iptrunk_speed,
subscription.iptrunk.iptrunk_sides[side_index].iptrunk_side_ae_members,
......
......@@ -26,14 +26,8 @@ disable_error_code = "annotation-unchecked"
enable_error_code = "ignore-without-code"
[tool.ruff]
exclude = [
".git",
".*_cache",
".tox",
"*.egg-info",
"__pycache__",
extend-exclude = [
"htmlcov",
"venv",
"gso/migrations",
"docs",
]
......@@ -42,7 +36,8 @@ ignore = [
"D213",
"N805",
"PLR0913",
"PLR0904"
"PLR0904",
"PLW1514"
]
line-length = 120
select = [
......
......@@ -186,7 +186,7 @@ def data_config_filename(configuration_data) -> str:
yield f.name
del os.environ["OSS_PARAMS_FILENAME"]
os.remove(f.name)
Path(f.name).unlink()
@pytest.fixture(scope="session")
......
......@@ -8,7 +8,7 @@ from gso.schedules.scheduling import scheduler
@pytest.fixture(scope="module")
def validate_subscriptions():
from gso.schedules.validate_subscriptions import validate_subscriptions as vs
from gso.schedules.validate_subscriptions import validate_subscriptions as vs # noqa: PLC0415
return vs
......@@ -32,7 +32,7 @@ def mock_logger():
yield mock
@pytest.fixture
@pytest.fixture()
def mock_celery():
with patch("gso.schedules.scheduling.current_app") as mock_app:
yield mock_app
......@@ -85,7 +85,9 @@ def test_no_subscriptions(mock_get_insync_subscriptions, mock_logger, validate_s
def test_subscriptions_without_system_target_workflow(
mock_get_insync_subscriptions, mock_logger, validate_subscriptions
mock_get_insync_subscriptions,
mock_logger,
validate_subscriptions,
):
mock_get_insync_subscriptions.return_value = [MagicMock(product=MagicMock(workflows=[]))]
validate_subscriptions()
......@@ -93,7 +95,9 @@ def test_subscriptions_without_system_target_workflow(
def test_subscription_status_not_usable(
mock_get_insync_subscriptions, mock_get_execution_context, validate_subscriptions
mock_get_insync_subscriptions,
mock_get_execution_context,
validate_subscriptions,
):
subscription_mock = MagicMock()
subscription_mock.product.workflows = [MagicMock(target=Target.SYSTEM, name="workflow_name")]
......@@ -107,7 +111,9 @@ def test_subscription_status_not_usable(
def test_valid_subscriptions_for_validation(
mock_get_insync_subscriptions, mock_get_execution_context, validate_subscriptions
mock_get_insync_subscriptions,
mock_get_execution_context,
validate_subscriptions,
):
subscription_mock = MagicMock()
mocked_workflow = MagicMock(target=Target.SYSTEM, name="workflow_name")
......@@ -117,5 +123,6 @@ def test_valid_subscriptions_for_validation(
validate_subscriptions()
validate_func = mock_get_execution_context()["validate"]
validate_func.assert_called_once_with(
mocked_workflow.name, json=[{"subscription_id": str(subscription_mock.subscription_id)}]
mocked_workflow.name,
json=[{"subscription_id": str(subscription_mock.subscription_id)}],
)
0% Loading or .
You are about to add 0 people to the discussion. Proceed with caution.
Please register or to comment