20122680412

Committed 11 Dec 2025 05:07AM UTC coverage: 59.066% (-0.3%) from 59.326%

Build # 20122680412

Build Type

push

github

Committed by

popstas

Commit Message

feat: validate single-word folder names and show warning

Run Details

7 of 23 new or added lines in 2 files covered. (30.43%)

19 existing lines in 4 files now uncovered.

1417 of 2399 relevant lines covered (59.07%)

0.59 hits per line

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

49.53

/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()
                folder_name_words = folder_name.split()
                name_warning = ""
                if len(folder_name_words) == 1:
                    name_warning = " \n" + self.translate("folder_name_single_word")
                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']}"
                    )
                    answer_text = (
                        self.translate("client_folder_exists", folder_url=folder_url)
                        + name_warning
                    )
                    self.send_json(
                        200,
                        {"answer": answer_text},
                    )
                    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']}"
            answer_text = (
                self.translate(
                    "folder_created",
                    folder_name=folder_name,
                    details=answer,
                    folder_url=folder_url,
                )
                + name_warning
            )
            self.send_json(
                200,
                {
                    "answer": answer_text,
                    "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	folder_name_words = folder_name.split()	1✔
136	name_warning = ""	1✔
137	if len(folder_name_words) == 1:	1✔
138	name_warning = " \n" + self.translate("folder_name_single_word")	1✔
139	task_id, initial_assignee_ids = get_task_and_assignees(	1✔
140	planfix_client, contact_id
141	)
142	folder, created = create_client_folder(	1✔
143	service, drive_config, folder_name
144	)
145	if not created:	1✔
146	folder_url = (	1✔
147	f"https://drive.google.com/drive/folders/{folder['id']}"
148	)
149	answer_text = (	1✔
150	self.translate("client_folder_exists", folder_url=folder_url)
151	+ name_warning
152	)
153	self.send_json(	1✔
154	200,
155	{"answer": answer_text},
156	)
157	return	1✔
158
159	access_report = grant_access(	×
160	planfix_client,
161	service,
162	drive_config,
163	role,
164	task_id,
165	initial_assignee_ids,
166	folder["id"],
167	)
168	except LocalizedError as exc:	×
169	self.send_json(200, {"answer": self.translate(exc.key, **exc.context)})	×
170	return	×
171	except Exception as exc: # pylint: disable=broad-except	×
172	logger.exception("Failed to process request: {}", exc)	×
173	self.send_json(200, {"answer": self.translate("internal_server_error")})	×
174	return	×
175
176	granted_accounts = access_report["granted_accounts"]	×
177	existing_accounts = access_report["existing_accounts"]	×
178	answer = self.translate(	×
179	"granted_existing",
180	granted=self._format_accounts(granted_accounts),
181	existing=self._format_accounts(existing_accounts),
182	)
183	folder_url = f"https://drive.google.com/drive/folders/{folder['id']}"	×
NEW 184	answer_text = (	×
185	self.translate(
186	"folder_created",
187	folder_name=folder_name,
188	details=answer,
189	folder_url=folder_url,
190	)
191	+ name_warning
192	)
UNCOV 193	self.send_json(	×
194	200,
195	{
196	"answer": answer_text,
197	"folder_id": folder["id"],
198	"folder_url": folder_url,
199	"granted_accounts": granted_accounts,
200	"existing_accounts": existing_accounts,
201	},
202	)
203
204	def do_POST(self) -> None: # noqa: N802	1✔
205	if self.path == "/set_client_folder_access":	×
206	if not self.authenticate():	×
207	return	×
208
209	try:	×
210	payload = self.parse_json_body()	×
211	except LocalizedError as exc:	×
212	self.send_json(	×
213	200, {"answer": self.translate(exc.key, **exc.context)}
214	)
215	return	×
216
217	self._log_request(payload)	×
218	self._handle_set_client_folder_access(payload)	×
219	return	×
220
221	if self.path == "/create_client_folder":	×
222	if not self.authenticate():	×
223	return	×
224
225	try:	×
226	payload = self.parse_json_body()	×
227	except LocalizedError as exc:	×
228	self.send_json(	×
229	200, {"answer": self.translate(exc.key, **exc.context)}
230	)
231	return	×
232
233	self._log_request(payload)	×
234	self._handle_create_client_folder(payload)	×
235	return	×
236
237	self.send_json(200, {"answer": translate(language, "not_found")})	×
238
239	return AccessHandler	1✔

popstas / google-drive-access / 20122680412

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