7891177570

pending completion

Build # 7891177570

Build Type

Pull #26

github

Committed by

web-flow

Commit Message

Merge 81b0d312c into ce6ca99dc

Pull Request Pull Request #26: Order programs

Run Details

0 of 1 new or added line in 1 file covered. (0.0%)

246 of 348 relevant lines covered (70.69%)

0.71 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

85.19

/winterapi/fidelius.py

"""
Fidelius is the keeper of secrets. It is the class that handles the secrets.
"""

import logging

import numpy as np
from filelock import FileLock
from wintertoo.models import Program

from winterapi.credentials import (
    TIMEOUT,
    WinterAPICredentials,
    clear_credentials_cache,
    get_secrets,
    secrets_lock_path,
    write_secrets,
)

logger = logging.getLogger(__name__)


class Fidelius:
    """
    Class to handle the secrets.
    """

    def __init__(self):
        self.credentials = self.load_secrets()

    @staticmethod
    def load_secrets() -> WinterAPICredentials:
        """
        Load the secrets from the keyring.

        :return:
        """
        secret_dict = get_secrets()
        return WinterAPICredentials(**secret_dict)

    def reload_secrets(self):
        """
        Reload the secrets from the keyring.

        :return: None
        """
        self.credentials = self.load_secrets()

    def export_secrets(self):
        """
        Export the secrets to the keyring.

        :return: None
        """
        write_secrets(secrets_dict=self.credentials.dict())

    def set_user(self, user: str, password: str, overwrite=False):
        """
        Function to set the user and password.

        :param user: User name.
        :param password: Password.
        :param overwrite: bool, whether to overwrite existing user/password.
        :return: None
        """
        self.reload_secrets()
        with FileLock(secrets_lock_path, timeout=TIMEOUT):
            if np.logical_and(self.credentials.user is None, not overwrite):
                err = f"User/password already set, and overwrite is set to {overwrite}."
                logger.error(err)
                raise ValueError(err)

            self.credentials.user = user
            self.credentials.password = password
            self.export_secrets()

    def get_user(self) -> str:
        """
        Get the user name.

        :return: user name
        """
        if self.credentials.user is None:
            err = "No user has been set"
            logger.error(err)
            raise KeyError(err)

        return self.credentials.user

    def get_password(self) -> str:
        """
        Get the password.

        :return: Password
        """
        if self.credentials.password is None:
            err = "No password has been set"
            logger.error(err)
            raise KeyError(err)
        return self.credentials.password

    def add_program(self, program_details: dict, overwrite=False):
        """
        Function to add a program to the credentials.

        :param program_details: Program details.
        :param overwrite: Bool, whether to overwrite existing program.
        :return: None
        """
        self.reload_secrets()

        program_details.pop("puid")

        program_details = Program(**program_details)

        with FileLock(secrets_lock_path, timeout=TIMEOUT):
            if np.logical_and(
                program_details.progname in self.credentials.programs, not overwrite
            ):
                err = (
                    f"Program {program_details.progname} already set, "
                    f"and overwrite is set to {overwrite}."
                )
                logger.error(err)
                raise ValueError(err)

            self.credentials.programs[program_details.progname] = program_details
            self.export_secrets()

    def get_programs(self) -> list[Program]:
        """
        Function to get the list of programs.

        :return: List of programs.
        """
        return sorted(list(self.credentials.programs))

    def get_all_program_details(self) -> dict[str, Program]:
        """
        Return all the program details.

        :return: Program details.
        """
        return self.credentials.programs

    def get_program_details(self, program_name: str) -> Program:
        """
        Get the details of a program.

        :param program_name: Name of the program.
        :return: Program details.
        """
        if program_name not in self.credentials.programs:
            err = (
                f"Program {program_name} not found in"
                f" {list(self.credentials.programs)}"
            )
            logger.error(err)
            raise KeyError(err)
        return self.credentials.programs[program_name]

    def delete_program(self, program_name: str):
        """
        Function to delete a program.

        :param program_name: Name of the program.
        :return: None
        """
        self.reload_secrets()
        with FileLock(secrets_lock_path, timeout=TIMEOUT):
            assert self.get_program_details(program_name) is not None
            del self.credentials.programs[program_name]
            self.export_secrets()

    @staticmethod
    def clear_cache():
        """
        Clear the cache.

        :return: None
        """
        clear_credentials_cache()

1	"""
2	Fidelius is the keeper of secrets. It is the class that handles the secrets.
3	"""
4
5	import logging	2✔
6
7	import numpy as np	2✔
8	from filelock import FileLock	2✔
9	from wintertoo.models import Program	2✔
10
11	from winterapi.credentials import (	2✔
12	TIMEOUT,
13	WinterAPICredentials,
14	clear_credentials_cache,
15	get_secrets,
16	secrets_lock_path,
17	write_secrets,
18	)
19
20	logger = logging.getLogger(__name__)	2✔
21
22
23	class Fidelius:	2✔
24	"""
25	Class to handle the secrets.
26	"""
27
28	def __init__(self):	2✔
29	self.credentials = self.load_secrets()	2✔
30
31	@staticmethod	2✔
32	def load_secrets() -> WinterAPICredentials:	2✔
33	"""
34	Load the secrets from the keyring.
35
36	:return:
37	"""
38	secret_dict = get_secrets()	2✔
39	return WinterAPICredentials(**secret_dict)	2✔
40
41	def reload_secrets(self):	2✔
42	"""
43	Reload the secrets from the keyring.
44
45	:return: None
46	"""
47	self.credentials = self.load_secrets()	2✔
48
49	def export_secrets(self):	2✔
50	"""
51	Export the secrets to the keyring.
52
53	:return: None
54	"""
55	write_secrets(secrets_dict=self.credentials.dict())	2✔
56
57	def set_user(self, user: str, password: str, overwrite=False):	2✔
58	"""
59	Function to set the user and password.
60
61	:param user: User name.
62	:param password: Password.
63	:param overwrite: bool, whether to overwrite existing user/password.
64	:return: None
65	"""
66	self.reload_secrets()	2✔
67	with FileLock(secrets_lock_path, timeout=TIMEOUT):	2✔
68	if np.logical_and(self.credentials.user is None, not overwrite):	2✔
69	err = f"User/password already set, and overwrite is set to {overwrite}."
70	logger.error(err)
71	raise ValueError(err)
72
73	self.credentials.user = user	2✔
74	self.credentials.password = password	2✔
75	self.export_secrets()	2✔
76
77	def get_user(self) -> str:	2✔
78	"""
79	Get the user name.
80
81	:return: user name
82	"""
83	if self.credentials.user is None:	2✔
84	err = "No user has been set"
85	logger.error(err)
86	raise KeyError(err)
87
88	return self.credentials.user	2✔
89
90	def get_password(self) -> str:	2✔
91	"""
92	Get the password.
93
94	:return: Password
95	"""
96	if self.credentials.password is None:	2✔
97	err = "No password has been set"
98	logger.error(err)
99	raise KeyError(err)
100	return self.credentials.password	2✔
101
102	def add_program(self, program_details: dict, overwrite=False):	2✔
103	"""
104	Function to add a program to the credentials.
105
106	:param program_details: Program details.
107	:param overwrite: Bool, whether to overwrite existing program.
108	:return: None
109	"""
110	self.reload_secrets()	2✔
111
112	program_details.pop("puid")	2✔
113
114	program_details = Program(**program_details)	2✔
115
116	with FileLock(secrets_lock_path, timeout=TIMEOUT):	2✔
117	if np.logical_and(	2✔
118	program_details.progname in self.credentials.programs, not overwrite
119	):
120	err = (
121	f"Program {program_details.progname} already set, "
122	f"and overwrite is set to {overwrite}."
123	)
124	logger.error(err)
125	raise ValueError(err)
126
127	self.credentials.programs[program_details.progname] = program_details	2✔
128	self.export_secrets()	2✔
129
130	def get_programs(self) -> list[Program]:	2✔
131	"""
132	Function to get the list of programs.
133
134	:return: List of programs.
135	"""
NEW 136	return sorted(list(self.credentials.programs))	×
137
138	def get_all_program_details(self) -> dict[str, Program]:	2✔
139	"""
140	Return all the program details.
141
142	:return: Program details.
143	"""
144	return self.credentials.programs	×
145
146	def get_program_details(self, program_name: str) -> Program:	2✔
147	"""
148	Get the details of a program.
149
150	:param program_name: Name of the program.
151	:return: Program details.
152	"""
153	if program_name not in self.credentials.programs:	2✔
154	err = (
155	f"Program {program_name} not found in"
156	f" {list(self.credentials.programs)}"
157	)
158	logger.error(err)
159	raise KeyError(err)
160	return self.credentials.programs[program_name]	2✔
161
162	def delete_program(self, program_name: str):	2✔
163	"""
164	Function to delete a program.
165
166	:param program_name: Name of the program.
167	:return: None
168	"""
169	self.reload_secrets()	×
170	with FileLock(secrets_lock_path, timeout=TIMEOUT):	×
171	assert self.get_program_details(program_name) is not None	×
172	del self.credentials.programs[program_name]	×
173	self.export_secrets()	×
174
175	@staticmethod	2✔
176	def clear_cache():	2✔
177	"""
178	Clear the cache.
179
180	:return: None
181	"""
182	clear_credentials_cache()	×

winter-telescope / winterapi / 7891177570

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous