19750033005

Committed 27 Nov 2025 11:38PM UTC coverage: 77.269% (+0.05%) from 77.221%

Build # 19750033005

Build Type

push

github

Committed by

popstas

Commit Message

fix: remove obsolete test for deleted get_log_level function

Run Details

1013 of 1311 relevant lines covered (77.27%)

0.77 hits per line

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

47.52

/src/drive_audit/http_handler.py

"""HTTP handler for managing Google Drive access."""

import json
from typing import Any, Dict, List

from loguru import logger

from .access_service import (
    create_client_folder,
    extract_folder_id,
    get_task_and_assignees,
    grant_access,
    normalize_assignee_ids,
    parse_assignee_ids,
)
from .http_utils import JsonRequestHandler, LocalizedError
from .model import DriveConfig, HttpConfig
from .planfix_client import PlanfixClient
from .translations import translate


def create_handler(
    planfix_client: PlanfixClient,
    service,
    http_config: HttpConfig,
    drive_config: DriveConfig,
    role: str,
):
    handler_http_config = http_config
    handler_language = http_config.lang

    class AccessHandler(JsonRequestHandler):
        http_config = handler_http_config
        language = handler_language

        def _log_request(self, payload: Dict[str, Any]) -> None:
            logger.info(
                "{} request: {}", self.path, json.dumps(payload, ensure_ascii=False)
            )

        def _format_accounts(self, accounts: List[str]) -> str:
            return ", ".join(accounts) if accounts else self.translate("none")

        def _handle_set_client_folder_access(self, payload: Dict[str, Any]) -> None:
            required_fields = ["contact_id", "folder_url"]
            missing_fields = [
                field for field in required_fields if field not in payload
            ]
            if missing_fields:
                self.send_json(
                    200,
                    {
                        "answer": self.translate(
                            "missing_fields", fields=", ".join(missing_fields)
                        )
                    },
                )
                return

            try:
                contact_id = int(payload["contact_id"])
                folder_id = extract_folder_id(str(payload["folder_url"]))

                has_task_id = "task_id" in payload
                has_assignee_id = "assignee_id" in payload

                if has_task_id and has_assignee_id:
                    task_id = int(payload["task_id"])
                    initial_assignee_ids = normalize_assignee_ids(
                        parse_assignee_ids(payload["assignee_id"])
                    )
                elif not has_task_id and not has_assignee_id:
                    task_id, initial_assignee_ids = get_task_and_assignees(
                        planfix_client, contact_id
                    )
                else:
                    self.send_json(
                        200,
                        {"answer": self.translate("task_and_assignee_together")},
                    )
                    return

                access_report = grant_access(
                    planfix_client,
                    service,
                    drive_config,
                    role,
                    task_id,
                    initial_assignee_ids,
                    folder_id,
                )
            except LocalizedError as exc:
                self.send_json(200, {"answer": self.translate(exc.key, **exc.context)})
                return
            except Exception as exc:  # pylint: disable=broad-except
                logger.exception("Failed to process request: {}", exc)
                self.send_json(200, {"answer": self.translate("internal_server_error")})
                return

            granted_accounts = access_report["granted_accounts"]
            existing_accounts = access_report["existing_accounts"]
            answer = self.translate(
                "granted_existing",
                granted=self._format_accounts(granted_accounts),
                existing=self._format_accounts(existing_accounts),
            )
            self.send_json(
                200,
                {
                    "answer": answer,
                    "granted_accounts": granted_accounts,
                    "existing_accounts": existing_accounts,
                },
            )

        def _handle_create_client_folder(self, payload: Dict[str, Any]) -> None:
            required_fields = ["contact_id", "folder_name"]
            missing_fields = [
                field for field in required_fields if field not in payload
            ]
            if missing_fields:
                self.send_json(
                    200,
                    {
                        "answer": self.translate(
                            "missing_fields", fields=", ".join(missing_fields)
                        )
                    },
                )
                return

            try:
                contact_id = int(payload["contact_id"])
                folder_name = str(payload["folder_name"]).strip()
                task_id, initial_assignee_ids = get_task_and_assignees(
                    planfix_client, contact_id
                )
                folder, created = create_client_folder(
                    service, drive_config, folder_name
                )
                if not created:
                    folder_url = (
                        f"https://drive.google.com/drive/folders/{folder['id']}"
                    )
                    self.send_json(
                        200,
                        {
                            "answer": self.translate(
                                "client_folder_exists", folder_url=folder_url
                            )
                        },
                    )
                    return

                access_report = grant_access(
                    planfix_client,
                    service,
                    drive_config,
                    role,
                    task_id,
                    initial_assignee_ids,
                    folder["id"],
                )
            except LocalizedError as exc:
                self.send_json(200, {"answer": self.translate(exc.key, **exc.context)})
                return
            except Exception as exc:  # pylint: disable=broad-except
                logger.exception("Failed to process request: {}", exc)
                self.send_json(200, {"answer": self.translate("internal_server_error")})
                return

            granted_accounts = access_report["granted_accounts"]
            existing_accounts = access_report["existing_accounts"]
            answer = self.translate(
                "granted_existing",
                granted=self._format_accounts(granted_accounts),
                existing=self._format_accounts(existing_accounts),
            )
            folder_url = f"https://drive.google.com/drive/folders/{folder['id']}"
            self.send_json(
                200,
                {
                    "answer": self.translate(
                        "folder_created",
                        folder_name=folder_name,
                        details=answer,
                        folder_url=folder_url,
                    ),
                    "folder_id": folder["id"],
                    "folder_url": folder_url,
                    "granted_accounts": granted_accounts,
                    "existing_accounts": existing_accounts,
                },
            )

        def do_POST(self) -> None:  # noqa: N802
            if self.path == "/set_client_folder_access":
                if not self.authenticate():
                    return

                try:
                    payload = self.parse_json_body()
                except LocalizedError as exc:
                    self.send_json(
                        200, {"answer": self.translate(exc.key, **exc.context)}
                    )
                    return

                self._log_request(payload)
                self._handle_set_client_folder_access(payload)
                return

            if self.path == "/create_client_folder":
                if not self.authenticate():
                    return

                try:
                    payload = self.parse_json_body()
                except LocalizedError as exc:
                    self.send_json(
                        200, {"answer": self.translate(exc.key, **exc.context)}
                    )
                    return

                self._log_request(payload)
                self._handle_create_client_folder(payload)
                return

            self.send_json(200, {"answer": translate(language, "not_found")})

    return AccessHandler

1	"""HTTP handler for managing Google Drive access."""
2
3	import json	1✔
4	from typing import Any, Dict, List	1✔
5
6	from loguru import logger	1✔
7
8	from .access_service import (	1✔
9	create_client_folder,
10	extract_folder_id,
11	get_task_and_assignees,
12	grant_access,
13	normalize_assignee_ids,
14	parse_assignee_ids,
15	)
16	from .http_utils import JsonRequestHandler, LocalizedError	1✔
17	from .model import DriveConfig, HttpConfig	1✔
18	from .planfix_client import PlanfixClient	1✔
19	from .translations import translate	1✔
20
21
22	def create_handler(	1✔
23	planfix_client: PlanfixClient,
24	service,
25	http_config: HttpConfig,
26	drive_config: DriveConfig,
27	role: str,
28	):
29	handler_http_config = http_config	1✔
30	handler_language = http_config.lang	1✔
31
32	class AccessHandler(JsonRequestHandler):	1✔
33	http_config = handler_http_config	1✔
34	language = handler_language	1✔
35
36	def _log_request(self, payload: Dict[str, Any]) -> None:	1✔
37	logger.info(	×
38	"{} request: {}", self.path, json.dumps(payload, ensure_ascii=False)
39	)
40
41	def _format_accounts(self, accounts: List[str]) -> str:	1✔
42	return ", ".join(accounts) if accounts else self.translate("none")	×
43
44	def _handle_set_client_folder_access(self, payload: Dict[str, Any]) -> None:	1✔
45	required_fields = ["contact_id", "folder_url"]	1✔
46	missing_fields = [	1✔
47	field for field in required_fields if field not in payload
48	]
49	if missing_fields:	1✔
50	self.send_json(	1✔
51	200,
52	{
53	"answer": self.translate(
54	"missing_fields", fields=", ".join(missing_fields)
55	)
56	},
57	)
58	return	1✔
59
60	try:	1✔
61	contact_id = int(payload["contact_id"])	1✔
62	folder_id = extract_folder_id(str(payload["folder_url"]))	1✔
63
64	has_task_id = "task_id" in payload	1✔
65	has_assignee_id = "assignee_id" in payload	1✔
66
67	if has_task_id and has_assignee_id:	1✔
68	task_id = int(payload["task_id"])	1✔
69	initial_assignee_ids = normalize_assignee_ids(	1✔
70	parse_assignee_ids(payload["assignee_id"])
71	)
72	elif not has_task_id and not has_assignee_id:	×
73	task_id, initial_assignee_ids = get_task_and_assignees(	×
74	planfix_client, contact_id
75	)
76	else:
77	self.send_json(	×
78	200,
79	{"answer": self.translate("task_and_assignee_together")},
80	)
81	return	×
82
83	access_report = grant_access(	×
84	planfix_client,
85	service,
86	drive_config,
87	role,
88	task_id,
89	initial_assignee_ids,
90	folder_id,
91	)
92	except LocalizedError as exc:	1✔
93	self.send_json(200, {"answer": self.translate(exc.key, **exc.context)})	1✔
94	return	1✔
95	except Exception as exc: # pylint: disable=broad-except	×
96	logger.exception("Failed to process request: {}", exc)	×
97	self.send_json(200, {"answer": self.translate("internal_server_error")})	×
98	return	×
99
100	granted_accounts = access_report["granted_accounts"]	×
101	existing_accounts = access_report["existing_accounts"]	×
102	answer = self.translate(	×
103	"granted_existing",
104	granted=self._format_accounts(granted_accounts),
105	existing=self._format_accounts(existing_accounts),
106	)
107	self.send_json(	×
108	200,
109	{
110	"answer": answer,
111	"granted_accounts": granted_accounts,
112	"existing_accounts": existing_accounts,
113	},
114	)
115
116	def _handle_create_client_folder(self, payload: Dict[str, Any]) -> None:	1✔
117	required_fields = ["contact_id", "folder_name"]	1✔
118	missing_fields = [	1✔
119	field for field in required_fields if field not in payload
120	]
121	if missing_fields:	1✔
122	self.send_json(	×
123	200,
124	{
125	"answer": self.translate(
126	"missing_fields", fields=", ".join(missing_fields)
127	)
128	},
129	)
130	return	×
131
132	try:	1✔
133	contact_id = int(payload["contact_id"])	1✔
134	folder_name = str(payload["folder_name"]).strip()	1✔
135	task_id, initial_assignee_ids = get_task_and_assignees(	1✔
136	planfix_client, contact_id
137	)
138	folder, created = create_client_folder(	1✔
139	service, drive_config, folder_name
140	)
141	if not created:	1✔
142	folder_url = (	1✔
143	f"https://drive.google.com/drive/folders/{folder['id']}"
144	)
145	self.send_json(	1✔
146	200,
147	{
148	"answer": self.translate(
149	"client_folder_exists", folder_url=folder_url
150	)
151	},
152	)
153	return	1✔
154
155	access_report = grant_access(	×
156	planfix_client,
157	service,
158	drive_config,
159	role,
160	task_id,
161	initial_assignee_ids,
162	folder["id"],
163	)
164	except LocalizedError as exc:	×
165	self.send_json(200, {"answer": self.translate(exc.key, **exc.context)})	×
166	return	×
167	except Exception as exc: # pylint: disable=broad-except	×
168	logger.exception("Failed to process request: {}", exc)	×
169	self.send_json(200, {"answer": self.translate("internal_server_error")})	×
170	return	×
171
172	granted_accounts = access_report["granted_accounts"]	×
173	existing_accounts = access_report["existing_accounts"]	×
174	answer = self.translate(	×
175	"granted_existing",
176	granted=self._format_accounts(granted_accounts),
177	existing=self._format_accounts(existing_accounts),
178	)
179	folder_url = f"https://drive.google.com/drive/folders/{folder['id']}"	×
180	self.send_json(	×
181	200,
182	{
183	"answer": self.translate(
184	"folder_created",
185	folder_name=folder_name,
186	details=answer,
187	folder_url=folder_url,
188	),
189	"folder_id": folder["id"],
190	"folder_url": folder_url,
191	"granted_accounts": granted_accounts,
192	"existing_accounts": existing_accounts,
193	},
194	)
195
196	def do_POST(self) -> None: # noqa: N802	1✔
197	if self.path == "/set_client_folder_access":	×
198	if not self.authenticate():	×
199	return	×
200
201	try:	×
202	payload = self.parse_json_body()	×
203	except LocalizedError as exc:	×
204	self.send_json(	×
205	200, {"answer": self.translate(exc.key, **exc.context)}
206	)
207	return	×
208
209	self._log_request(payload)	×
210	self._handle_set_client_folder_access(payload)	×
211	return	×
212
213	if self.path == "/create_client_folder":	×
214	if not self.authenticate():	×
215	return	×
216
217	try:	×
218	payload = self.parse_json_body()	×
219	except LocalizedError as exc:	×
220	self.send_json(	×
221	200, {"answer": self.translate(exc.key, **exc.context)}
222	)
223	return	×
224
225	self._log_request(payload)	×
226	self._handle_create_client_folder(payload)	×
227	return	×
228
229	self.send_json(200, {"answer": translate(language, "not_found")})	×
230
231	return AccessHandler	1✔

popstas / google-drive-access / 19750033005

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