diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml
index 70bfc27e17dce4bdd2df10ee86b2490ec7789d98..45164592a74d3f5e59204397932331edde5141a3 100644
--- a/.pre-commit-config.yaml
+++ b/.pre-commit-config.yaml
@@ -9,7 +9,9 @@ repos:
- --fix
- --preview
- --ignore=PLR0917,PLR0914
+ - --extend-exclude=test/*
# Run the formatter.
- id: ruff-format
args:
- --preview
+ - --exclude=test/*
diff --git a/Changelog.md b/Changelog.md
index e5624c2a65a0a385f73ba3774b9734479c6db862..d87116de079b184e123062fd106f19bcd2abc6c9 100644
--- a/Changelog.md
+++ b/Changelog.md
@@ -1,6 +1,8 @@
# Changelog
All notable changes to this project will be documented in this file.
+## [0.7] - 2024-02-21
+- Infoblox client: added support for the `network_view` (IPAM).
## [0.6] - 2024-02-07
- Removed the import workflows from migrations.
diff --git a/docs/source/glossary.rst b/docs/source/glossary.rst
index 959d89761d27f3652cbae2d840a5e2ef2f7a3703..0bbcce276f79bee067aef114ea5b0f0c3101ab39 100644
--- a/docs/source/glossary.rst
+++ b/docs/source/glossary.rst
@@ -58,6 +58,9 @@ Glossary of terms
Simple Network Management Protocol: a protocol that's used for gathering data, widely used for network management
and monitoring.
+ TWAMP
+ Two-way Active Measurement Protocol.
+
UUID
Universally Unique Identifier
diff --git a/docs/vale/styles/config/vocabularies/geant-jargon/accept.txt b/docs/vale/styles/config/vocabularies/geant-jargon/accept.txt
index 081f7913cba3d95518c76bfb75278bdf9300dbf8..9172c42e39a10632a75a6e73c17dd9b0e9b5eb41 100644
--- a/docs/vale/styles/config/vocabularies/geant-jargon/accept.txt
+++ b/docs/vale/styles/config/vocabularies/geant-jargon/accept.txt
@@ -14,4 +14,6 @@ Dark_fiber
PHASE 1
[Mm]odify
AAI
-[M|m]iddleware
\ No newline at end of file
+[M|m]iddleware
+TWAMP
+Pydantic
diff --git a/gso/auth/oidc_policy_helper.py b/gso/auth/oidc_policy_helper.py
index 04e2fc8e5ec419fb5f80a6a5646bb3429512cad9..241641dcfe3a93ffd41bb2962255c6557b95d235 100644
--- a/gso/auth/oidc_policy_helper.py
+++ b/gso/auth/oidc_policy_helper.py
@@ -82,13 +82,8 @@ class OIDCUserModel(dict):
if the attribute is one of the registered claims or raises an AttributeError
if the key is not found.
- Args:
- ----
- key: The attribute name to retrieve.
-
- Returns:
- -------
- The value of the attribute if it exists, otherwise raises AttributeError.
+ :param str key: The attribute name to retrieve.
+ :return: The value of the attribute if it exists, otherwise raises AttributeError.
"""
try:
return object.__getattribute__(self, key)
@@ -167,6 +162,7 @@ class OPAResult(BaseModel):
----------
- result (bool): Indicates whether the access request is allowed or denied.
- decision_id (str): A unique identifier for the decision made by OPA.
+
"""
result: bool = False
@@ -208,15 +204,10 @@ class OIDCUser(HTTPBearer):
This is used as a security module in Fastapi projects
- Args:
- ----
- request: Starlette request method.
- token: Optional value to directly pass a token.
-
- Returns:
- -------
- OIDCUserModel object.
+ :param Request request: Starlette request method.
+ :param str token: Optional value to directly pass a token.
+ :return: OIDCUserModel object.
"""
if not oauth2lib_settings.OAUTH2_ACTIVE:
return None
@@ -380,16 +371,12 @@ def opa_decision(
to authorize requests based on OPA policies. It utilizes OIDC for user information and makes a
call to the OPA service to determine authorization.
- Args:
- ----
- opa_url: URL of the Open Policy Agent service.
- oidc_security: An instance of OIDCUser for user authentication.
- auto_error: If True, automatically raises an HTTPException on authorization failure.
- opa_kwargs: Additional keyword arguments to be passed to the OPA input.
+ :param str opa_url: URL of the Open Policy Agent service.
+ :param OIDCUser oidc_security: An instance of OIDCUser for user authentication.
+ :param bool auto_error: If True, automatically raises an HTTPException on authorization failure.
+ :param Mapping[str, str] | None opa_kwargs: Additional keyword arguments to be passed to the OPA input.
- Returns:
- -------
- An asynchronous decision function that can be used as a dependency in FastAPI endpoints.
+ :return: An asynchronous decision function that can be used as a dependency in FastAPI endpoints.
"""
async def _opa_decision(
@@ -407,6 +394,7 @@ def opa_decision(
request: Request object that will be used to retrieve request metadata.
user_info: The OIDCUserModel object that will be checked
async_request: The :term:`httpx` client.
+
"""
if not (oauth2lib_settings.OAUTH2_ACTIVE and oauth2lib_settings.OAUTH2_AUTHORIZATION_ACTIVE):
return None
diff --git a/gso/auth/security.py b/gso/auth/security.py
index 16065e467e02176d92df20563c4c3e0f56845667..e1d5376479e9e95f50847af06fa7120272135a4a 100644
--- a/gso/auth/security.py
+++ b/gso/auth/security.py
@@ -34,8 +34,6 @@ def get_oidc_user() -> OIDCUser:
This function returns the instance of OIDCUser initialized in the module.
It is typically used for accessing the OIDCUser across different parts of the application.
- Returns
- -------
- OIDCUser: The instance of OIDCUser configured with OAuth2 settings.
+ :return OIDCUser: The instance of OIDCUser configured with OAuth2 settings.
"""
return oidc_user
diff --git a/gso/oss-params-example.json b/gso/oss-params-example.json
index 931367ea48746610b29646651e99a68f89d1288e..2ef1cd14749760e423c59676896c32638286989b 100644
--- a/gso/oss-params-example.json
+++ b/gso/oss-params-example.json
@@ -18,31 +18,36 @@
"V4": {"containers": [], "networks": ["1.1.0.0/24"], "mask": 0},
"V6": {"containers": [], "networks": ["dead:beef::/64"], "mask": 0},
"domain_name": ".lo",
- "dns_view": "default"
+ "dns_view": "default",
+ "network_view": "default"
},
"TRUNK": {
"V4": {"containers": ["1.1.1.0/24"], "networks": [], "mask": 31},
"V6": {"containers": ["dead:beef::/64"], "networks": [], "mask": 126},
"domain_name": ".trunk",
- "dns_view": "default"
+ "dns_view": "default",
+ "network_view": "default"
},
"GEANT_IP": {
"V4": {"containers": ["1.1.2.0/24"], "networks": [], "mask": 31},
"V6": {"containers": ["dead:beef::/64"], "networks": [], "mask": 126},
"domain_name": ".geantip",
- "dns_view": "default"
+ "dns_view": "default",
+ "network_view": "default"
},
"SI": {
"V4": {"containers": ["1.1.3.0/24"], "networks": [], "mask": 31},
"V6": {"containers": ["dead:beef::/64"], "networks": [], "mask": 126},
"domain_name": ".si",
- "dns_view": "default"
+ "dns_view": "default",
+ "network_view": "default"
},
"LT_IAS": {
"V4": {"containers": ["1.1.4.0/24"], "networks": [], "mask": 31},
"V6": {"containers": ["dead:beef::/64"], "networks": [], "mask": 126},
"domain_name": ".ltias",
- "dns_view": "default"
+ "dns_view": "default",
+ "network_view": "default"
}
},
"MONITORING": {
@@ -77,5 +82,13 @@
"THIRD_PARTY_API_KEYS": {
"AnsibleDynamicInventoryGenerator": "REALLY_random_AND_secure_T0keN",
"Application_2": "another_REALY_random_AND_3cure_T0keN"
+ },
+ "EMAIL": {
+ "from_address": "noreply@nren.local",
+ "smtp_host": "smtp.nren.local",
+ "smtp_port": 487,
+ "starttls_enabled": true,
+ "smtp_username": "username",
+ "smtp_password": "password"
}
}
diff --git a/gso/products/product_blocks/router.py b/gso/products/product_blocks/router.py
index d8e9996ba0821d48765ae36f50df1bb2aa541752..5d1a307d85825759b734a0c06b665d1e7eb06866 100644
--- a/gso/products/product_blocks/router.py
+++ b/gso/products/product_blocks/router.py
@@ -68,12 +68,12 @@ class RouterBlockProvisioning(RouterBlockInactive, lifecycle=[SubscriptionLifecy
router_fqdn: str
router_ts_port: PortNumber
- router_access_via_ts: bool | None = None
- router_lo_ipv4_address: ipaddress.IPv4Address | None = None
- router_lo_ipv6_address: ipaddress.IPv6Address | None = None
- router_lo_iso_address: str | None = None
- router_role: RouterRole | None = None
- router_site: SiteBlockProvisioning | None
+ router_access_via_ts: bool
+ router_lo_ipv4_address: ipaddress.IPv4Address
+ router_lo_ipv6_address: ipaddress.IPv6Address
+ router_lo_iso_address: str
+ router_role: RouterRole
+ router_site: SiteBlockProvisioning
vendor: RouterVendor
diff --git a/gso/products/product_blocks/site.py b/gso/products/product_blocks/site.py
index efe21c4b270a8974c1615b89f025012ff8793582..1852b24615076b2d76dde41db71a9e5d5fcc535f 100644
--- a/gso/products/product_blocks/site.py
+++ b/gso/products/product_blocks/site.py
@@ -82,16 +82,16 @@ class SiteBlockInactive(
class SiteBlockProvisioning(SiteBlockInactive, lifecycle=[SubscriptionLifecycle.PROVISIONING]):
"""A site that's currently being provisioned, see :class:`SiteBlock`."""
- site_name: str | None = None
- site_city: str | None = None
- site_country: str | None = None
- site_country_code: str | None = None
- site_latitude: LatitudeCoordinate | None = None
- site_longitude: LongitudeCoordinate | None = None
- site_internal_id: int | None = None
- site_bgp_community_id: int | None = None
- site_tier: SiteTier | None = None
- site_ts_address: str | None = None
+ site_name: str
+ site_city: str
+ site_country: str
+ site_country_code: str
+ site_latitude: LatitudeCoordinate
+ site_longitude: LongitudeCoordinate
+ site_internal_id: int
+ site_bgp_community_id: int
+ site_tier: SiteTier
+ site_ts_address: str
class SiteBlock(SiteBlockProvisioning, lifecycle=[SubscriptionLifecycle.ACTIVE]):
@@ -120,4 +120,4 @@ class SiteBlock(SiteBlockProvisioning, lifecycle=[SubscriptionLifecycle.ACTIVE])
#: The address of the terminal server that this router is connected to. The terminal server provides out of band
#: access. This is required in case a link goes down, or when a router is initially added to the network and it
#: does not have any IP trunks connected to it.
- site_ts_address: str | None = None
+ site_ts_address: str
diff --git a/gso/services/infoblox.py b/gso/services/infoblox.py
index 0c7176deb5dd41b6a1c6d7cf5dcea30a2992a4ba..6838ad9eada39ad5fe3999f0f28096e04216388a 100644
--- a/gso/services/infoblox.py
+++ b/gso/services/infoblox.py
@@ -41,29 +41,32 @@ def _setup_connection() -> tuple[connector.Connector, IPAMParams]:
def _allocate_network(
- conn: connector.Connector, dns_view: str, netmask: int, containers: list[str], comment: str | None = ""
+ conn: connector.Connector,
+ dns_view: str,
+ network_view: str,
+ netmask: int,
+ containers: list[str],
+ comment: str | None = "",
) -> ipaddress.IPv4Network | ipaddress.IPv6Network:
"""Allocate a new network in Infoblox.
The function will go over all given containers, and try to allocate a network within the available IP space. If no
space is available, this method raises an :class:`AllocationError`.
- :param conn: An active Infoblox connection.
- :type conn: :class:`infoblox_client.connector.Connector`
- :param dns_view: The Infoblox ``dns_view`` in which the network should be allocated.
- :type dns_view: str
- :param netmask: The netmask of the desired network. Can be up to 32 for v4 networks, and 128 for v6 networks.
- :type netmask: int
- :param containers: A list of network containers in which the network should be allocated, given in :term:`CIDR`
- notation.
- :type containers: list[str]
- :param comment: Optionally, a comment can be added to the network allocation.
- :type comment: str, optional
+ :param :class:`infoblox_client.connector.Connector` conn: An active Infoblox connection.
+ :param str dns_view: The Infoblox ``dns_view`` in which the network should be allocated.
+ :param str network_view: The Infoblox ``network_view`` where the network should be allocated.
+ :param int netmask: The netmask of the desired network. Can be up to 32 for v4 networks, and 128 for v6 networks.
+ :param list [str] containers: A list of network containers in which the network should be allocated, given in
+ :term:`CIDR` notation.
+ :param str comment: Optionally, a comment can be added to the network allocation.
"""
for container in [ipaddress.ip_network(con) for con in containers]:
for network in container.subnets(new_prefix=netmask):
if objects.Network.search(conn, network=str(network)) is None:
- created_net = objects.Network.create(conn, network=str(network), dns_view=dns_view, comment=comment)
+ created_net = objects.Network.create(
+ conn, network=str(network), dns_view=dns_view, network_view=network_view, comment=comment
+ )
if created_net.response != "Infoblox Object already Exists":
return ipaddress.ip_network(created_net.network)
msg = f"IP container {container} appears to be full."
@@ -104,8 +107,9 @@ def allocate_v4_network(service_type: str, comment: str | None = "") -> ipaddres
netmask = getattr(oss, service_type).V4.mask
containers = getattr(oss, service_type).V4.containers
dns_view = getattr(oss, service_type).dns_view
+ network_view = getattr(oss, service_type).network_view
- return ipaddress.IPv4Network(_allocate_network(conn, dns_view, netmask, containers, comment))
+ return ipaddress.IPv4Network(_allocate_network(conn, dns_view, network_view, netmask, containers, comment))
def allocate_v6_network(service_type: str, comment: str | None = "") -> ipaddress.IPv6Network:
@@ -123,8 +127,9 @@ def allocate_v6_network(service_type: str, comment: str | None = "") -> ipaddres
netmask = getattr(oss, service_type).V6.mask
containers = getattr(oss, service_type).V6.containers
dns_view = getattr(oss, service_type).dns_view
+ network_view = getattr(oss, service_type).network_view
- return ipaddress.IPv6Network(_allocate_network(conn, dns_view, netmask, containers, comment))
+ return ipaddress.IPv6Network(_allocate_network(conn, dns_view, network_view, netmask, containers, comment))
def find_network_by_cidr(
@@ -184,10 +189,11 @@ def allocate_host(
allocation_networks_v4 = getattr(oss, service_type).V4.networks
allocation_networks_v6 = getattr(oss, service_type).V6.networks
dns_view = getattr(oss, service_type).dns_view
+ network_view = getattr(oss, service_type).network_view
created_v6 = None
for ipv6_range in allocation_networks_v6:
- v6_alloc = objects.IPAllocation.next_available_ip_from_cidr(dns_view, str(ipv6_range))
+ v6_alloc = objects.IPAllocation.next_available_ip_from_cidr(network_view, str(ipv6_range))
ipv6_object = objects.IP.create(ip=v6_alloc, mac=NULL_MAC, configure_for_dhcp=False)
try:
new_host = objects.HostRecord.create(
@@ -197,6 +203,7 @@ def allocate_host(
aliases=cname_aliases,
comment=comment,
dns_view=dns_view,
+ network_view=network_view,
)
created_v6 = ipaddress.IPv6Address(new_host.ipv6addr)
except InfobloxCannotCreateObject:
@@ -209,7 +216,7 @@ def allocate_host(
created_v4 = None
for ipv4_range in allocation_networks_v4:
- v4_alloc = objects.IPAllocation.next_available_ip_from_cidr(dns_view, str(ipv4_range))
+ v4_alloc = objects.IPAllocation.next_available_ip_from_cidr(network_view, str(ipv4_range))
ipv4_object = objects.IP.create(ip=v4_alloc, mac=NULL_MAC, configure_for_dhcp=False)
new_host = objects.HostRecord.search(conn, name=hostname)
new_host.ipv4addrs = [ipv4_object]
@@ -240,7 +247,8 @@ def create_host_by_ip(
:param str hostname: The :term:`FQDN` of the new host.
:param IPv4Address ipv4_address: The IPv4 address of the new host.
:param IPv6Address ipv6_address: The IPv6 address of the new host.
- :param str service_type: The relevant service type, used to deduce the correct ``dns_view`` in Infoblox.
+ :param str service_type: The relevant service type, used to deduce the correct ``dns_view`` and ``network_view`` in
+ Infoblox.
:param str comment: The comment stored in this Infoblox record, most likely the relevant ``subscription_id`` in
:term:`GSO`.
"""
@@ -252,9 +260,12 @@ def create_host_by_ip(
ipv6_object = objects.IP.create(ip=ipv6_address, mac=NULL_MAC, configure_for_dhcp=False)
ipv4_object = objects.IP.create(ip=ipv4_address, mac=NULL_MAC, configure_for_dhcp=False)
dns_view = getattr(oss, service_type).dns_view
+ network_view = getattr(oss, service_type).network_view
# This needs to be done in two steps, otherwise only one of the IP addresses is stored.
- objects.HostRecord.create(conn, ip=ipv6_object, name=hostname, comment=comment, dns_view=dns_view)
+ objects.HostRecord.create(
+ conn, ip=ipv6_object, name=hostname, comment=comment, dns_view=dns_view, network_view=network_view
+ )
new_host = find_host_by_fqdn(hostname)
new_host.ipv4addrs = [ipv4_object]
new_host.update()
diff --git a/gso/services/mailer.py b/gso/services/mailer.py
new file mode 100644
index 0000000000000000000000000000000000000000..3cd85b7369fd176d55d02674f9b55e2c02a47a72
--- /dev/null
+++ b/gso/services/mailer.py
@@ -0,0 +1,32 @@
+"""The mailer service sends notification emails, as part of workflows that require interaction with external parties."""
+
+import smtplib
+from email.message import EmailMessage
+from ssl import create_default_context
+
+from gso.settings import load_oss_params
+
+
+def send_mail(recipient: str, subject: str, body: str) -> None:
+ """Send an email message to the given address.
+
+ Only supports STARTTLS, not SSL.
+
+ :param str recipient: The destination address.
+ :param str subject: The email subject.
+ :param str body: The contents of the email message.
+ """
+ email_params = load_oss_params().EMAIL
+ msg = EmailMessage()
+ msg["From"] = email_params.from_address
+ msg["To"] = recipient
+ msg["Subject"] = subject
+ msg.set_content(body)
+
+ with smtplib.SMTP(email_params.smtp_host, email_params.smtp_port) as s:
+ if email_params.starttls_enabled:
+ tls_context = create_default_context()
+ s.starttls(context=tls_context)
+ if email_params.smtp_username and email_params.smtp_password:
+ s.login(email_params.smtp_username, email_params.smtp_password)
+ s.send_message(msg)
diff --git a/gso/settings.py b/gso/settings.py
index ca0da591c00e829585db55c5a68731b3c44aac90..2055e96c78e9177aad9f2b06b2871666b4eced70 100644
--- a/gso/settings.py
+++ b/gso/settings.py
@@ -81,6 +81,7 @@ class ServiceNetworkParams(BaseSettings):
V6: V6NetworkParams
domain_name: str
dns_view: str
+ network_view: str
class IPAMParams(BaseSettings):
@@ -152,6 +153,18 @@ class NetBoxParams(BaseSettings):
api: str
+class EmailParams(BaseSettings):
+ """Parameters for the email service."""
+
+ # TODO: Use more strict types after we've migrated to Pydantic 2.x
+ from_address: str
+ smtp_host: str
+ smtp_port: int
+ starttls_enabled: bool
+ smtp_username: str | None
+ smtp_password: str | None
+
+
class OSSParams(BaseSettings):
"""The set of parameters required for running :term:`GSO`."""
@@ -162,6 +175,7 @@ class OSSParams(BaseSettings):
PROVISIONING_PROXY: ProvisioningProxyParams
CELERY: CeleryParams
THIRD_PARTY_API_KEYS: dict[str, str]
+ EMAIL: EmailParams
def load_oss_params() -> OSSParams:
diff --git a/gso/utils/helpers.py b/gso/utils/helpers.py
index 689ce2c9e3dd262ec328afb3ed0a8a20abcebf87..c43a80737f56e37e3288e78d89a0ab38dced94e4 100644
--- a/gso/utils/helpers.py
+++ b/gso/utils/helpers.py
@@ -77,8 +77,7 @@ def available_interfaces_choices_including_current_members(
],
)
options = {
- interface["name"]: f"{interface['name']} - {interface['module']['display']} - {interface['description']}"
- for interface in available_interfaces
+ interface["name"]: f"{interface['name']} {interface['description']}" for interface in available_interfaces
}
return Choice("ae member", zip(options.keys(), options.items(), strict=True)) # type: ignore[arg-type]
@@ -271,3 +270,23 @@ def validate_interface_name_list(interface_name_list: list, vendor: str) -> list
)
raise ValueError(error_msg)
return interface_name_list
+
+
+def validate_tt_number(tt_number: str) -> str:
+ """Validate a string to match a specific pattern.
+
+ This method checks if the input string starts with 'TT#' and is followed by exactly 16 digits.
+
+ :param str tt_number: The TT number as string to validate
+
+ :return str: The TT number string if TT number match was successful, otherwise it will raise a ValueError.
+ """
+ pattern = r"^TT#\d{16}$"
+ if not bool(re.match(pattern, tt_number)):
+ err_msg = (
+ f"The given TT number: {tt_number} is not valid. "
+ f" A valid TT number starts with 'TT#' followed by 16 digits."
+ )
+ raise ValueError(err_msg)
+
+ return tt_number
diff --git a/gso/workflows/iptrunk/create_iptrunk.py b/gso/workflows/iptrunk/create_iptrunk.py
index acbfe3757568c2257e828e4ddbd506fcaac25bfd..db6f91ce99d706d1c97d59bbda15785532f36d50 100644
--- a/gso/workflows/iptrunk/create_iptrunk.py
+++ b/gso/workflows/iptrunk/create_iptrunk.py
@@ -17,12 +17,12 @@ from pynetbox.models.dcim import Interfaces
from gso.products.product_blocks.iptrunk import (
IptrunkInterfaceBlockInactive,
- IptrunkSideBlockProvisioning,
+ IptrunkSideBlockInactive,
IptrunkType,
PhyPortCapacity,
)
from gso.products.product_blocks.router import RouterVendor
-from gso.products.product_types.iptrunk import IptrunkInactive, IptrunkProvisioning
+from gso.products.product_types.iptrunk import IptrunkInactive
from gso.products.product_types.router import Router
from gso.services import infoblox, subscriptions
from gso.services.crm import get_customer_by_name
@@ -36,6 +36,7 @@ from gso.utils.helpers import (
validate_interface_name_list,
validate_iptrunk_unique_interface,
validate_router_in_netbox,
+ validate_tt_number,
)
@@ -57,6 +58,10 @@ def initial_input_form_generator(product_name: str) -> FormGenerator:
iptrunk_speed: PhyPortCapacity
iptrunk_minimum_links: int
+ @validator("tt_number", allow_reuse=True)
+ def validate_tt_number(cls, tt_number: str) -> str:
+ return validate_tt_number(tt_number)
+
initial_user_input = yield CreateIptrunkForm
router_enum_a = Choice("Select a router", zip(routers.keys(), routers.items(), strict=True)) # type: ignore[arg-type]
@@ -101,13 +106,11 @@ def initial_input_form_generator(product_name: str) -> FormGenerator:
side_a_ae_members: ae_members_side_a # type: ignore[valid-type]
@validator("side_a_ae_members", allow_reuse=True)
- def validate_iptrunk_unique_interface_side_a(cls, side_a_ae_members: list[LAGMember]) -> list[LAGMember]:
- return validate_iptrunk_unique_interface(side_a_ae_members)
-
- @validator("side_a_ae_members", allow_reuse=True)
- def validate_interface_name_members(cls, side_a_ae_members: list[LAGMember]) -> list[LAGMember]:
+ def validate_side_a_ae_members(cls, side_a_ae_members: list[LAGMember]) -> list[LAGMember]:
+ validate_iptrunk_unique_interface(side_a_ae_members)
vendor = get_router_vendor(router_a)
- return validate_interface_name_list(side_a_ae_members, vendor)
+ validate_interface_name_list(side_a_ae_members, vendor)
+ return side_a_ae_members
user_input_side_a = yield CreateIptrunkSideAForm
# Remove the selected router for side A, to prevent any loops
@@ -153,13 +156,11 @@ def initial_input_form_generator(product_name: str) -> FormGenerator:
side_b_ae_members: ae_members_side_b # type: ignore[valid-type]
@validator("side_b_ae_members", allow_reuse=True)
- def validate_iptrunk_unique_interface_side_b(cls, side_b_ae_members: list[LAGMember]) -> list[LAGMember]:
- return validate_iptrunk_unique_interface(side_b_ae_members)
-
- @validator("side_b_ae_members", allow_reuse=True)
- def validate_interface_name_members(cls, side_b_ae_members: list[LAGMember]) -> list[LAGMember]:
+ def validate_side_b_ae_members(cls, side_b_ae_members: list[LAGMember]) -> list[LAGMember]:
+ validate_iptrunk_unique_interface(side_b_ae_members)
vendor = get_router_vendor(router_b)
- return validate_interface_name_list(side_b_ae_members, vendor)
+ validate_interface_name_list(side_b_ae_members, vendor)
+ return side_b_ae_members
user_input_side_b = yield CreateIptrunkSideBForm
@@ -184,7 +185,7 @@ def create_subscription(product: UUIDstr, customer: str) -> State:
@step("Get information from IPAM")
-def get_info_from_ipam(subscription: IptrunkProvisioning) -> State:
+def get_info_from_ipam(subscription: IptrunkInactive) -> State:
"""Allocate IP resources in :term:`IPAM`."""
subscription.iptrunk.iptrunk_ipv4_network = infoblox.allocate_v4_network(
"TRUNK",
@@ -242,14 +243,13 @@ def initialize_subscription(
)
side_names = sorted([side_a.router_site.site_name, side_b.router_site.site_name])
subscription.description = f"IP trunk {side_names[0]} {side_names[1]}, geant_s_sid:{geant_s_sid}"
- subscription = IptrunkProvisioning.from_other_lifecycle(subscription, SubscriptionLifecycle.PROVISIONING)
return {"subscription": subscription}
@step("Provision IP trunk interface [DRY RUN]")
def provision_ip_trunk_iface_dry(
- subscription: IptrunkProvisioning,
+ subscription: IptrunkInactive,
callback_route: str,
process_id: UUIDstr,
tt_number: str,
@@ -277,7 +277,7 @@ def provision_ip_trunk_iface_dry(
@step("Provision IP trunk interface [FOR REAL]")
def provision_ip_trunk_iface_real(
- subscription: IptrunkProvisioning,
+ subscription: IptrunkInactive,
callback_route: str,
process_id: UUIDstr,
tt_number: str,
@@ -305,7 +305,7 @@ def provision_ip_trunk_iface_real(
@step("Check IP connectivity of the trunk")
def check_ip_trunk_connectivity(
- subscription: IptrunkProvisioning,
+ subscription: IptrunkInactive,
callback_route: str,
) -> State:
"""Check successful connectivity across the new trunk."""
@@ -323,7 +323,7 @@ def check_ip_trunk_connectivity(
@step("Provision IP trunk ISIS interface [DRY RUN]")
def provision_ip_trunk_isis_iface_dry(
- subscription: IptrunkProvisioning,
+ subscription: IptrunkInactive,
callback_route: str,
process_id: UUIDstr,
tt_number: str,
@@ -351,7 +351,7 @@ def provision_ip_trunk_isis_iface_dry(
@step("Provision IP trunk ISIS interface [FOR REAL]")
def provision_ip_trunk_isis_iface_real(
- subscription: IptrunkProvisioning,
+ subscription: IptrunkInactive,
callback_route: str,
process_id: UUIDstr,
tt_number: str,
@@ -379,7 +379,7 @@ def provision_ip_trunk_isis_iface_real(
@step("Check ISIS adjacency")
def check_ip_trunk_isis(
- subscription: IptrunkProvisioning,
+ subscription: IptrunkInactive,
callback_route: str,
) -> State:
"""Run an Ansible playbook to confirm :term:`ISIS` adjacency."""
@@ -396,7 +396,7 @@ def check_ip_trunk_isis(
@step("NextBox integration")
-def reserve_interfaces_in_netbox(subscription: IptrunkProvisioning) -> State:
+def reserve_interfaces_in_netbox(subscription: IptrunkInactive) -> State:
"""Create the :term:`LAG` interfaces in NetBox and attach the lag interfaces to the physical interfaces."""
nbclient = NetboxClient()
for trunk_side in subscription.iptrunk.iptrunk_sides:
@@ -428,22 +428,25 @@ def reserve_interfaces_in_netbox(subscription: IptrunkProvisioning) -> State:
}
-def _allocate_interfaces_in_netbox(iptrunk_side: IptrunkSideBlockProvisioning) -> None:
+def _allocate_interfaces_in_netbox(iptrunk_side: IptrunkSideBlockInactive) -> None:
for interface in iptrunk_side.iptrunk_side_ae_members:
- NetboxClient().allocate_interface(
- device_name=iptrunk_side.iptrunk_side_node.router_fqdn,
- iface_name=interface.interface_name,
- )
+ fqdn = iptrunk_side.iptrunk_side_node.router_fqdn
+ iface_name = interface.interface_name
+ if not fqdn or not iface_name:
+ msg = f"FQDN and/or interface name missing in subscription {interface.owner_subscription_id}"
+ raise ValueError(msg)
+
+ NetboxClient().allocate_interface(device_name=fqdn, iface_name=iface_name)
@step("Allocate interfaces in Netbox for side A")
-def netbox_allocate_side_a_interfaces(subscription: IptrunkProvisioning) -> None:
+def netbox_allocate_side_a_interfaces(subscription: IptrunkInactive) -> None:
"""Allocate the :term:`LAG` interfaces for the Nokia router on side A."""
_allocate_interfaces_in_netbox(subscription.iptrunk.iptrunk_sides[0])
@step("Allocate interfaces in Netbox for side B")
-def netbox_allocate_side_b_interfaces(subscription: IptrunkProvisioning) -> None:
+def netbox_allocate_side_b_interfaces(subscription: IptrunkInactive) -> None:
"""Allocate the :term:`LAG` interfaces for the Nokia router on side B."""
_allocate_interfaces_in_netbox(subscription.iptrunk.iptrunk_sides[1])
@@ -484,7 +487,7 @@ def create_iptrunk() -> StepList:
>> pp_interaction(check_ip_trunk_isis)
>> side_a_is_nokia(netbox_allocate_side_a_interfaces)
>> side_b_is_nokia(netbox_allocate_side_b_interfaces)
- >> set_status(SubscriptionLifecycle.ACTIVE)
+ >> set_status(SubscriptionLifecycle.PROVISIONING)
>> resync
>> done
)
diff --git a/gso/workflows/iptrunk/deploy_twamp.py b/gso/workflows/iptrunk/deploy_twamp.py
index c5b74be3909bf073fa38e416555c5c781df73901..8e2e1016b17fbfc6ba5840e8f47a9c831c9ced17 100644
--- a/gso/workflows/iptrunk/deploy_twamp.py
+++ b/gso/workflows/iptrunk/deploy_twamp.py
@@ -7,9 +7,11 @@ from orchestrator.types import FormGenerator, State, UUIDstr
from orchestrator.workflow import StepList, done, init, step, workflow
from orchestrator.workflows.steps import resync, store_process_subscription, unsync
from orchestrator.workflows.utils import wrap_modify_initial_input_form
+from pydantic import validator
from gso.products.product_types.iptrunk import Iptrunk
from gso.services.provisioning_proxy import execute_playbook, pp_interaction
+from gso.utils.helpers import validate_tt_number
def _initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
@@ -23,6 +25,10 @@ def _initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
)
tt_number: str
+ @validator("tt_number", allow_reuse=True)
+ def validate_tt_number(cls, tt_number: str) -> str:
+ return validate_tt_number(tt_number)
+
user_input = yield DeployTWAMPForm
return user_input.dict()
diff --git a/gso/workflows/iptrunk/migrate_iptrunk.py b/gso/workflows/iptrunk/migrate_iptrunk.py
index 810041cd89a7a5c7104ddd8a8ab5f368ac5ac068..71729149bf86db868b0bfb477565120ff8e4d98d 100644
--- a/gso/workflows/iptrunk/migrate_iptrunk.py
+++ b/gso/workflows/iptrunk/migrate_iptrunk.py
@@ -39,6 +39,7 @@ from gso.utils.helpers import (
available_lags_choices,
get_router_vendor,
validate_interface_name_list,
+ validate_tt_number,
)
from gso.utils.workflow_steps import set_isis_to_90000
@@ -71,6 +72,10 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
migrate_to_different_site: bool = False
restore_isis_metric: bool = True
+ @validator("tt_number", allow_reuse=True, pre=True, always=True)
+ def validate_tt_number(cls, tt_number: str) -> str:
+ return validate_tt_number(tt_number)
+
migrate_form_input = yield IPTrunkMigrateForm
current_routers = [
diff --git a/gso/workflows/iptrunk/modify_trunk_interface.py b/gso/workflows/iptrunk/modify_trunk_interface.py
index 8394170f7d983da26d125ba3d0a95a749937d49e..1dcda4fd5ceb2f3b2a4114533ff313a76823ac53 100644
--- a/gso/workflows/iptrunk/modify_trunk_interface.py
+++ b/gso/workflows/iptrunk/modify_trunk_interface.py
@@ -32,6 +32,7 @@ from gso.utils.helpers import (
get_router_vendor,
validate_interface_name_list,
validate_iptrunk_unique_interface,
+ validate_tt_number,
)
@@ -91,6 +92,10 @@ def initial_input_form_generator(subscription_id: UUIDstr) -> FormGenerator:
iptrunk_ipv4_network: ipaddress.IPv4Network = ReadOnlyField(subscription.iptrunk.iptrunk_ipv4_network)
iptrunk_ipv6_network: ipaddress.IPv6Network = ReadOnlyField(subscription.iptrunk.iptrunk_ipv6_network)
+ @validator("tt_number", allow_reuse=True)
+ def validate_tt_number(cls, tt_number: str) -> str:
+ return validate_tt_number(tt_number)
+
initial_user_input = yield ModifyIptrunkForm
ae_members_side_a = initialize_ae_members(subscription, initial_user_input.dict(), 0)
diff --git a/gso/workflows/iptrunk/terminate_iptrunk.py b/gso/workflows/iptrunk/terminate_iptrunk.py
index cd2614c7027e531cea5e1aa482e5b65c439cbb46..33ee86fde6129a05278d8985a3959e2ef945c067 100644
--- a/gso/workflows/iptrunk/terminate_iptrunk.py
+++ b/gso/workflows/iptrunk/terminate_iptrunk.py
@@ -16,6 +16,7 @@ from orchestrator.workflows.steps import (
unsync,
)
from orchestrator.workflows.utils import wrap_modify_initial_input_form
+from pydantic import validator
from gso.products.product_blocks.iptrunk import IptrunkSideBlock
from gso.products.product_blocks.router import RouterVendor
@@ -23,7 +24,7 @@ from gso.products.product_types.iptrunk import Iptrunk
from gso.services import infoblox
from gso.services.netbox_client import NetboxClient
from gso.services.provisioning_proxy import execute_playbook, pp_interaction
-from gso.utils.helpers import get_router_vendor
+from gso.utils.helpers import get_router_vendor, validate_tt_number
from gso.utils.workflow_steps import set_isis_to_90000
@@ -40,6 +41,10 @@ def initial_input_form_generator() -> FormGenerator:
clean_up_ipam: bool = True
clean_up_netbox: bool = True
+ @validator("tt_number", allow_reuse=True)
+ def validate_tt_number(cls, tt_number: str) -> str:
+ return validate_tt_number(tt_number)
+
user_input = yield TerminateForm
return user_input.dict()
diff --git a/gso/workflows/router/create_router.py b/gso/workflows/router/create_router.py
index 6bec84a6a1eca393da95d68e28258e3c70d1946c..29a60dbb87d73f68f695710d3b746a8e8fa4c198 100644
--- a/gso/workflows/router/create_router.py
+++ b/gso/workflows/router/create_router.py
@@ -2,11 +2,12 @@
from typing import Any
+from orchestrator.config.assignee import Assignee
from orchestrator.forms import FormPage
-from orchestrator.forms.validators import Choice
+from orchestrator.forms.validators import Choice, Label
from orchestrator.targets import Target
from orchestrator.types import FormGenerator, State, SubscriptionLifecycle, UUIDstr
-from orchestrator.workflow import StepList, conditional, done, init, step, workflow
+from orchestrator.workflow import StepList, conditional, done, init, inputstep, step, workflow
from orchestrator.workflows.steps import resync, set_status, store_process_subscription
from orchestrator.workflows.utils import wrap_create_initial_input_form
from pydantic import validator
@@ -18,7 +19,7 @@ from gso.products.product_blocks.router import (
RouterVendor,
generate_fqdn,
)
-from gso.products.product_types.router import RouterInactive, RouterProvisioning
+from gso.products.product_types.router import RouterInactive
from gso.products.product_types.site import Site
from gso.services import infoblox, subscriptions
from gso.services.crm import get_customer_by_name
@@ -106,15 +107,16 @@ def initialize_subscription(
subscription.router.vendor = vendor
subscription.description = f"Router {fqdn}"
- subscription = RouterProvisioning.from_other_lifecycle(subscription, SubscriptionLifecycle.PROVISIONING)
-
return {"subscription": subscription}
@step("Allocate loopback interfaces in IPAM")
-def ipam_allocate_loopback(subscription: RouterProvisioning) -> State:
+def ipam_allocate_loopback(subscription: RouterInactive) -> State:
"""Allocate :term:`IPAM` resources for the loopback interface."""
fqdn = subscription.router.router_fqdn
+ if not fqdn:
+ msg = f"Router fqdn for subscription id {subscription.subscription_id} is missing!"
+ raise ValueError(msg)
loopback_v4, loopback_v6 = infoblox.allocate_host(f"lo0.{fqdn}", "LO", [fqdn], str(subscription.subscription_id))
subscription.router.router_lo_ipv4_address = loopback_v4
@@ -125,17 +127,21 @@ def ipam_allocate_loopback(subscription: RouterProvisioning) -> State:
@step("Create NetBox Device")
-def create_netbox_device(subscription: RouterProvisioning) -> State:
+def create_netbox_device(subscription: RouterInactive) -> State:
"""Create a new NOKIA device in Netbox."""
- NetboxClient().create_device(
- subscription.router.router_fqdn,
- str(subscription.router.router_site.site_tier), # type: ignore[union-attr]
- )
+ fqdn = subscription.router.router_fqdn
+ site_tier = subscription.router.router_site.site_tier if subscription.router.router_site else None
+ if not fqdn or not site_tier:
+ msg = f"FQDN and/or Site tier missing in router subscription {subscription.subscription_id}!"
+ raise ValueError(msg)
+
+ NetboxClient().create_device(fqdn, site_tier)
+
return {"subscription": subscription}
@step("Verify IPAM resources for loopback interface")
-def verify_ipam_loopback(subscription: RouterProvisioning) -> State:
+def verify_ipam_loopback(subscription: RouterInactive) -> State:
"""Validate the :term:`IPAM` resources for the loopback interface."""
host_record = infoblox.find_host_by_fqdn(f"lo0.{subscription.router.router_fqdn}")
if not host_record or str(subscription.subscription_id) not in host_record.comment:
@@ -144,6 +150,63 @@ def verify_ipam_loopback(subscription: RouterProvisioning) -> State:
return {"subscription": subscription}
+@inputstep("Prompt to reboot", assignee=Assignee.SYSTEM)
+def prompt_reboot_router(subscription: RouterInactive) -> FormGenerator:
+ """Wait for confirmation from an operator that the router has been rebooted."""
+
+ class RebootPrompt(FormPage):
+ class Config:
+ title = "Please reboot before continuing"
+
+ if subscription.router.router_site and subscription.router.router_site.site_ts_address:
+ info_label_1: Label = (
+ f"Base config has been deployed. Please log in via the console using https://" # type: ignore[assignment]
+ f"{subscription.router.router_site.site_ts_address}."
+ )
+ else:
+ info_label_1 = "Base config has been deployed. Please log in via the console." # type: ignore[assignment]
+
+ info_label_2: Label = "Reboot the router, and once it is up again, press submit to continue the workflow." # type: ignore[assignment]
+
+ yield RebootPrompt
+
+ return {}
+
+
+@inputstep("Prompt to test the console", assignee=Assignee.SYSTEM)
+def prompt_console_login() -> FormGenerator:
+ """Wait for confirmation from an operator that the router can be logged into."""
+
+ class ConsolePrompt(FormPage):
+ class Config:
+ title = "Verify local authentication"
+
+ info_label_1: Label = (
+ "Verify that you are able to log in to the router via the console using the admin account." # type: ignore[assignment]
+ )
+ info_label_2: Label = "Once this is done, press submit to continue the workflow." # type: ignore[assignment]
+
+ yield ConsolePrompt
+
+ return {}
+
+
+@inputstep("Prompt IMS insertion", assignee=Assignee.SYSTEM)
+def prompt_insert_in_ims() -> FormGenerator:
+ """Wait for confirmation from an operator that the router has been inserted in IMS."""
+
+ class IMSPrompt(FormPage):
+ class Config:
+ title = "Update IMS mediation server"
+
+ info_label_1: Label = "Insert the router into IMS." # type: ignore[assignment]
+ info_label_2: Label = "Once this is done, press submit to continue the workflow." # type: ignore[assignment]
+
+ yield IMSPrompt
+
+ return {}
+
+
@workflow(
"Create router",
initial_input_form=wrap_create_initial_input_form(initial_input_form_generator),
@@ -169,9 +232,12 @@ def create_router() -> StepList:
>> pp_interaction(deploy_base_config_dry)
>> pp_interaction(deploy_base_config_real)
>> verify_ipam_loopback
+ >> prompt_reboot_router
+ >> prompt_console_login
+ >> prompt_insert_in_ims
>> router_is_nokia(create_netbox_device)
>> pp_interaction(run_checks_after_base_config)
- >> set_status(SubscriptionLifecycle.ACTIVE)
+ >> set_status(SubscriptionLifecycle.PROVISIONING)
>> resync
>> done
)
diff --git a/gso/workflows/router/update_ibgp_mesh.py b/gso/workflows/router/update_ibgp_mesh.py
index d2f6d8d66b4dcd95250c6bcbc87f15815f3d15a6..d8ba52a09b36735bd46404a091f7f408ad4f2f66 100644
--- a/gso/workflows/router/update_ibgp_mesh.py
+++ b/gso/workflows/router/update_ibgp_mesh.py
@@ -2,10 +2,12 @@
from typing import Any
+from orchestrator.config.assignee import Assignee
from orchestrator.forms import FormPage
+from orchestrator.forms.validators import Label
from orchestrator.targets import Target
from orchestrator.types import FormGenerator, State, UUIDstr
-from orchestrator.workflow import StepList, done, init, step, workflow
+from orchestrator.workflow import StepList, done, init, inputstep, step, workflow
from orchestrator.workflows.steps import resync, store_process_subscription, unsync
from orchestrator.workflows.utils import wrap_modify_initial_input_form
from pydantic import root_validator
@@ -188,6 +190,36 @@ def add_device_to_librenms(subscription: Router) -> State:
return {"librenms_device": librenms_result}
+@inputstep("Prompt RADIUS insertion", assignee=Assignee.SYSTEM)
+def prompt_insert_in_radius() -> FormGenerator:
+ """Wait for confirmation from an operator that the router has been inserted in RADIUS."""
+
+ class RADIUSPrompt(FormPage):
+ class Config:
+ title = "Please update RADIUS before continuing"
+
+ info_label: Label = "Insert the router into RADIUS, and continue the workflow once this has been completed." # type: ignore[assignment]
+
+ yield RADIUSPrompt
+
+ return {}
+
+
+@inputstep("Prompt RADIUS login", assignee=Assignee.SYSTEM)
+def prompt_radius_login() -> FormGenerator:
+ """Wait for confirmation from an operator that the router can be logged into using RADIUS."""
+
+ class RADIUSPrompt(FormPage):
+ class Config:
+ title = "Please check RADIUS before continuing"
+
+ info_label: Label = "Log in to the router using RADIUS, and continue the workflow when this was successful." # type: ignore[assignment]
+
+ yield RADIUSPrompt
+
+ return {}
+
+
@step("Update subscription model")
def update_subscription_model(subscription: Router) -> State:
"""Update the database model, such that it should not be reached via :term:`OOB` access anymore."""
@@ -221,6 +253,8 @@ def update_ibgp_mesh() -> StepList:
>> pp_interaction(add_all_pe_to_p_real)
>> pp_interaction(check_ibgp_session)
>> add_device_to_librenms
+ >> prompt_insert_in_radius
+ >> prompt_radius_login
>> update_subscription_model
>> resync
>> done
diff --git a/setup.py b/setup.py
index 85e9b36696de7be28dea8acf89e2fddaa74cbcba..cf34f158746312524a27348396cf48a46812080d 100644
--- a/setup.py
+++ b/setup.py
@@ -2,7 +2,7 @@ from setuptools import find_packages, setup
setup(
name="geant-service-orchestrator",
- version="0.6",
+ version="0.7",
author="GÉANT",
author_email="swd@geant.org",
description="GÉANT Service Orchestrator",
diff --git a/test/conftest.py b/test/conftest.py
index dd6e9d6d3fc3429f0ce0df33213de8dc1f195523..b0e8ddbd1ff3a99521d1540d92f2ad519f253131 100644
--- a/test/conftest.py
+++ b/test/conftest.py
@@ -120,6 +120,7 @@ def configuration_data() -> dict:
},
"domain_name": ".geant.net",
"dns_view": "default",
+ "network_view": "default",
},
"TRUNK": {
"V4": {
@@ -134,6 +135,7 @@ def configuration_data() -> dict:
},
"domain_name": ".trunk",
"dns_view": "default",
+ "network_view": "default",
},
"GEANT_IP": {
"V4": {
@@ -148,6 +150,7 @@ def configuration_data() -> dict:
},
"domain_name": ".geantip",
"dns_view": "default",
+ "network_view": "default",
},
"SI": {
"V4": {
@@ -158,6 +161,7 @@ def configuration_data() -> dict:
"V6": {"containers": [], "networks": [], "mask": 126},
"domain_name": ".geantip",
"dns_view": "default",
+ "network_view": "default",
},
"LT_IAS": {
"V4": {
@@ -172,6 +176,7 @@ def configuration_data() -> dict:
},
"domain_name": ".geantip",
"dns_view": "default",
+ "network_view": "default",
},
},
"MONITORING": {
@@ -208,6 +213,14 @@ def configuration_data() -> dict:
"AnsibleDynamicInventoryGenerator": "REALY_random_AND_3cure_T0keN",
"Application_2": "another_REALY_random_AND_3cure_T0keN",
},
+ "EMAIL": {
+ "from_address": "noreply@nren.local",
+ "smtp_host": "smtp.nren.local",
+ "smtp_port": 487,
+ "starttls_enabled": True,
+ "smtp_username": "username",
+ "smtp_password": "password",
+ },
}
diff --git a/test/utils/test_helpers.py b/test/utils/test_helpers.py
index d5fb22827da3cc207bde920585a72d1ff05ded89..4006c4ad50101e93b10ca16c802761612a244ea7 100644
--- a/test/utils/test_helpers.py
+++ b/test/utils/test_helpers.py
@@ -4,7 +4,7 @@ import pytest
from gso.products.product_blocks.iptrunk import IptrunkInterfaceBlock
from gso.products.product_blocks.router import RouterVendor
-from gso.utils.helpers import available_interfaces_choices_including_current_members
+from gso.utils.helpers import available_interfaces_choices_including_current_members, validate_tt_number
@pytest.fixture()
@@ -21,6 +21,18 @@ def mock_netbox_client():
yield mock
+@pytest.fixture()
+def generate_tt_numbers(faker, request):
+ """Generator for valid and invalid tt numbers."""
+ valid_count = request.param.get("valid", 0)
+ invalid_count = request.param.get("invalid", 0)
+
+ valid_data = [(faker.tt_number(), True) for _ in range(valid_count)]
+ invalid_data = [(faker.sentence(), False) for _ in range(invalid_count)]
+
+ return valid_data + invalid_data
+
+
def test_non_nokia_router_returns_none(mock_router, faker):
mock_router.from_subscription.return_value.router.vendor = RouterVendor.JUNIPER
result = available_interfaces_choices_including_current_members(faker.uuid4(), "10G", [])
@@ -62,3 +74,19 @@ def test_nokia_router_with_interfaces_returns_choice(mock_router, mock_netbox_cl
assert hasattr(result, "interface1")
assert hasattr(result, "interface2")
assert hasattr(result, "interface3")
+
+
+@pytest.mark.parametrize("generate_tt_numbers", [{"valid": 5, "invalid": 3}], indirect=True)
+def test_tt_number(generate_tt_numbers):
+ """Test different TT numbers"""
+ for tt_number, is_valid in generate_tt_numbers:
+ if is_valid:
+ assert validate_tt_number(tt_number) == tt_number
+ else:
+ err_msg = (
+ f"The given TT number: {tt_number} is not valid. "
+ f" A valid TT number starts with 'TT#' followed by 16 digits."
+ )
+
+ with pytest.raises(ValueError, match=err_msg):
+ validate_tt_number(tt_number)
diff --git a/test/workflows/__init__.py b/test/workflows/__init__.py
index 669fd75cf6b91057d0c327b0dbc111abb63e6f8f..5470858c105666ac8396b42649c65a02eeef8d0c 100644
--- a/test/workflows/__init__.py
+++ b/test/workflows/__init__.py
@@ -140,7 +140,7 @@ class WorkflowInstanceForTests(LazyWorkflowInstance):
This can be as simple as merely importing a workflow function. However, if it concerns a workflow generating
function, that function will be called with or without arguments as specified.
- Returns: A workflow function.
+ :return Workflow: A workflow function.
"""
self.workflow.name = self.name
return self.workflow
diff --git a/test/workflows/iptrunk/test_create_iptrunk.py b/test/workflows/iptrunk/test_create_iptrunk.py
index 59ecc85e9e5fc3901ebf519246c9ca6fdb0a95ec..c069618476f87b4beafcf4fbcb89f259fc043071 100644
--- a/test/workflows/iptrunk/test_create_iptrunk.py
+++ b/test/workflows/iptrunk/test_create_iptrunk.py
@@ -129,7 +129,7 @@ def test_successful_iptrunk_creation_with_standard_lso_result(
subscription.iptrunk.iptrunk_sides[1].iptrunk_side_node.router_site.site_name,
]
)
- assert subscription.status == "active"
+ assert subscription.status == "provisioning"
assert subscription.description == (
f"IP trunk {sorted_sides[0]} {sorted_sides[1]}, geant_s_sid:{input_form_wizard_data[0]['geant_s_sid']}"
)
diff --git a/test/workflows/router/test_create_router.py b/test/workflows/router/test_create_router.py
index 957c3bda41026db88a7d7fc70ce2efa556770774..ca012757153478de3ca4e50fa37b44daa9696595 100644
--- a/test/workflows/router/test_create_router.py
+++ b/test/workflows/router/test_create_router.py
@@ -7,11 +7,14 @@ from gso.products import ProductType, Site
from gso.products.product_blocks.router import RouterRole, RouterVendor
from gso.products.product_types.router import Router
from gso.services.subscriptions import get_product_id_by_name
+from test import USER_CONFIRM_EMPTY_FORM
from test.workflows import (
assert_complete,
assert_pp_interaction_failure,
assert_pp_interaction_success,
+ assert_suspended,
extract_state,
+ resume_workflow,
run_workflow,
)
@@ -80,15 +83,22 @@ def test_create_nokia_router_success(
name=mock_fqdn,
)
- for _ in range(3):
+ for _ in range(2):
result, step_log = assert_pp_interaction_success(result, process_stat, step_log)
+ # Handle three consecutive user input steps
+ for _ in range(3):
+ assert_suspended(result)
+ result, step_log = resume_workflow(process_stat, step_log, input_data=USER_CONFIRM_EMPTY_FORM)
+
+ result, step_log = assert_pp_interaction_success(result, process_stat, step_log)
+
assert_complete(result)
state = extract_state(result)
subscription = Router.from_subscription(subscription_id)
- assert subscription.status == "active"
+ assert subscription.status == "provisioning"
assert subscription.description == f"Router {mock_fqdn}"
assert mock_provision_router.call_count == 3
@@ -167,7 +177,7 @@ def test_create_nokia_router_lso_failure(
assert_pp_interaction_failure(result, process_stat, step_log)
- assert subscription.status == "provisioning"
+ assert subscription.status == "initial"
assert subscription.description == f"Router {mock_fqdn}"
assert mock_provision_router.call_count == 2
diff --git a/test/workflows/router/test_update_ibgp_mesh.py b/test/workflows/router/test_update_ibgp_mesh.py
index 14543066130d06d5d6f5d148b2d126a44d267adc..8c7da5d5d0abb2a4275f52e2c89cc2c52bc7f51e 100644
--- a/test/workflows/router/test_update_ibgp_mesh.py
+++ b/test/workflows/router/test_update_ibgp_mesh.py
@@ -6,7 +6,8 @@ from pydantic_forms.exceptions import FormValidationError
from gso.products import Iptrunk
from gso.products.product_blocks.router import RouterRole
-from test.workflows import assert_pp_interaction_success, extract_state, run_workflow
+from test import USER_CONFIRM_EMPTY_FORM
+from test.workflows import assert_pp_interaction_success, assert_suspended, extract_state, resume_workflow, run_workflow
@pytest.fixture()
@@ -33,6 +34,11 @@ def test_update_ibgp_mesh_success(
for _ in range(5):
result, step_log = assert_pp_interaction_success(result, process_stat, step_log)
+ # Handle two consecutive user input steps
+ for _ in range(2):
+ assert_suspended(result)
+ result, step_log = resume_workflow(process_stat, step_log, input_data=USER_CONFIRM_EMPTY_FORM)
+
state = extract_state(result)
assert mock_execute_playbook.call_count == 5