"""Manage configuration settings and credentials for connecting to tango gql services."""
import os
from typing import NamedTuple, TypedDict, Union
[docs]class Settings(NamedTuple):
"""Bundle service name and tango gql end point."""
service_name: str
tangogql: str
[docs]class CredentailsDict(TypedDict):
"""Bundle user username and password as single dictionary object."""
username: str
password: str
[docs]class Credentials(NamedTuple):
"""Bundle user username and password as single object."""
username: str
password: str
[docs] def asdict(self) -> CredentailsDict:
"""Return credentials as a dictionary.
:return: the credentials as a dictionary
"""
return {"username": self.username, "password": self.password}
[docs]class Environment(NamedTuple):
"""Stores environment settings as a collection into a single object."""
username: str
password: str
domain: str
kube_branch: str
telescope: str
kubehost: str
tango_bridge_ip: Union[str, None] = None
bypass_auth: bool = False
kube_namespace: Union[str, None] = None
[docs] def get_credentials(self) -> Union[Credentials, None]:
"""Generate a Credentials object if exists.
:return: the credentials if they exist otherwise return None.
"""
if self.bypass_auth:
return None
return Credentials(self.username, self.password)
[docs]def get_env() -> Environment:
"""Collect the relevant environment settings from host env.
The following environmental settings is relevant for remoting:
1. KUBE_HOST: the domain name hosting the k8 cluster (default k8s.skao.stfc)
2. BYPASS_AUTH: whether authentication must be skipped (default is not to skip)
3. TARANTA_USER: the taranta username needed for authentication on tangogql service
4. TARANTA_PASSWORD: the taranta password needed for authentication on tangogql service
5. DOMAIN: The type of environment in which SUT is deployed nl: branch/integration/staging
6. KUBE_BRANCH: The branch name (in case of branch domain) used for the k8 deployment
7. TANGO_BRIDGE_IP: an alternative raw ip address to use for connecting to k8 cluster
Note KUBE_BRANCH is needed only when the DOMAIN variable is set to branch. The branch value is
the name of the git branch used to deploy an instance of the SUT. If TANGO_BRIDGE_IP is given
then the DOMAIN and KUBE_BRANCH variables become irrelevant.
:return: The environment settings as a single Environment object
"""
kubehost = os.getenv("KUBE_HOST", "k8s.skao.stfc")
bypass_auth = os.getenv("BYPASS_AUTH")
bypass_auth = True if bypass_auth else False
if bypass_auth:
username = ""
password = ""
bypass_auth = True
else:
username = os.getenv("TARANTA_USER")
assert username, (
"Unable to find username for taranta authentication, did you set "
"TARANTA_USER env variable?"
)
password = os.getenv("TARANTA_PASSWORD")
assert password, (
"Unable to find password for taranta authentication, did you set "
"TARANTA_PASSWORD env variable?"
)
domain = os.getenv("DOMAIN")
kube_branch = os.getenv("KUBE_BRANCH")
tango_bridge_ip = os.getenv("TANGO_BRIDGE_IP")
kube_namespace = os.getenv("KUBE_NAMESPACE")
if any([tango_bridge_ip, kube_namespace]):
kube_branch = ""
telescope = ""
domain = ""
else:
assert (
domain
), "Unable to find domain variable, did you set DOMAIN env variable?"
if domain != "branch":
kube_branch = "" # ignore kube_branch if domain is not set to branch
assert (
kube_branch
), "Unable to find kube branch variable, did you set KUBE_BRANCH env variable?"
telescope = os.getenv("TEL")
assert (
telescope
), "Unable to find telescope variable, did you set TEL env variable?"
return Environment(
username,
password,
domain,
kube_branch,
telescope,
kubehost,
tango_bridge_ip,
bypass_auth,
kube_namespace,
)
[docs]def get_tango_gql_rest_url(settings: Settings, env: Environment) -> str:
"""Generate a tango gql rest service url from a given environment and tangogql settings.
:param settings: The settings used to name the tangogql service
:type settings: Settings
:param env: The environment values obtained from the host.
:type env: Environment
:return: A tangogql rest http endpoint
"""
url = _get_tango_gql_service_url(settings, env)
return f"http://{url}/db"
[docs]def get_tango_gql_ws_url(settings: Settings, env: Environment) -> str:
"""Generate a tango gql websocket service url from a given environment and tangogql settings.
:param settings: The settings used to name the tangogql service
:type settings: Settings
:param env: The environment values obtained from the host.
:type env: Environment
:return: A tangogql websocket endpoint
"""
url = _get_tango_gql_service_url(settings, env)
return f"ws://{url}/socket"
def _get_tango_gql_service_url(settings: Settings, env: Environment) -> str:
if env.tango_bridge_ip:
return env.tango_bridge_ip
if env.kube_namespace:
return f"{env.kubehost}/{env.kube_namespace}/{settings.service_name}"
namespace_url = f"ci-ska-skampi-{env.kube_branch}-{env.telescope}"
integration_url = f"integration-{env.telescope}"
staging_url = f"staging-{env.telescope}"
if env.domain == "branch":
return f"{env.kubehost}/{namespace_url}/{settings.service_name}"
if env.domain == "integration":
return f"{env.kubehost}/{integration_url}/{settings.service_name}"
if env.domain == "staging":
return f"{env.kubehost}/{staging_url}/{settings.service_name}"
return (
f"taranta-taranta-test-{env.telescope}-{env.kube_branch}/{settings.tangogql}/"
)