19720685265

Committed 26 Nov 2025 11:52PM UTC coverage: 53.576%. Remained the same

Build # 19720685265

Build Type

Pull #13

github

Committed by

popstas

Commit Message

chore: apply black and isort formatting

Pull Request Pull Request #13: chore: apply black and isort formatting

Run Details

84 of 186 new or added lines in 15 files covered. (45.16%)

28 existing lines in 9 files now uncovered.

824 of 1538 relevant lines covered (53.58%)

0.54 hits per line

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

96.83

/tests/test_google_client_cache.py

from __future__ import annotations

from pathlib import Path
from typing import Dict, List

from drive_audit.google_client import (
    ensure_public_subdir,
    list_folder_children,
    reset_list_folder_children_cache,
    set_list_folder_children_cache_dir,
)


class FakeRequest:
    def __init__(self, response: Dict):
        self.response = response

    def execute(self) -> Dict:
        return self.response


class FakeFilesResource:
    def __init__(self, pages: List[Dict]):
        self.pages = list(pages)
        self.calls = 0

    def list(self, **kwargs):
        response = (
            self.pages[self.calls] if self.calls < len(self.pages) else {"files": []}
        )
        self.calls += 1
        return FakeRequest(response)


class FakeService:
    def __init__(self, pages: List[Dict]):
        self.files_resource = FakeFilesResource(pages)

    def files(self):
        return self.files_resource


def test_list_folder_children_caches_results(tmp_path: Path):
    set_list_folder_children_cache_dir(tmp_path)
    reset_list_folder_children_cache()

    pages = [
        {
            "files": [
                {
                    "id": "1",
                    "name": "first",
                    "mimeType": "text/plain",
                    "parents": ["folder"],
                },
            ],
            "nextPageToken": "next",
        },
        {
            "files": [
                {
                    "id": "2",
                    "name": "second",
                    "mimeType": "text/plain",
                    "parents": ["folder"],
                }
            ]
        },
    ]
    service = FakeService(pages)

    first_pass = list(
        list_folder_children(
            service,
            folder_id="folder",
            drive_id="drive",
            page_size=10,
            cache_timeout_seconds=3600,
        )
    )

    assert [file["id"] for file in first_pass] == ["1", "2"]
    assert service.files_resource.calls == 2
    assert list(tmp_path.glob("*.json"))

    second_pass = list(
        list_folder_children(
            service,
            folder_id="folder",
            drive_id="drive",
            page_size=10,
            cache_timeout_seconds=3600,
        )
    )

    assert [file["id"] for file in second_pass] == ["1", "2"]
    assert service.files_resource.calls == 2


def test_list_folder_children_skips_cache_when_disabled(tmp_path: Path):
    set_list_folder_children_cache_dir(tmp_path)
    reset_list_folder_children_cache()

    first_pages = [
        {
            "files": [
                {
                    "id": "cached",
                    "name": "cached",
                    "mimeType": "text/plain",
                    "parents": ["folder"],
                }
            ]
        }
    ]
    first_pass = list(
        list_folder_children(
            FakeService(first_pages),
            folder_id="folder",
            drive_id="",
            page_size=5,
            cache_timeout_seconds=0,
        )
    )
    second_pages = [
        {
            "files": [
                {
                    "id": "cached",
                    "name": "cached",
                    "mimeType": "text/plain",
                    "parents": ["folder"],
                }
            ]
        }
    ]
    second_pass = list(
        list_folder_children(
            FakeService(second_pages),
            folder_id="folder",
            drive_id="",
            page_size=5,
            cache_timeout_seconds=0,
        )
    )

    assert [file["id"] for file in first_pass] == ["cached"]
    assert [file["id"] for file in second_pass] == ["cached"]
    assert not list(tmp_path.glob("*.json"))


def test_ensure_public_subdir_reuses_existing(monkeypatch):
    created: Dict[str, Dict] = {}
    ensured: list[str] = []

    def fake_find_child_folder(service, parent_id, name, drive_id):
        created["last_lookup"] = {
            "parent_id": parent_id,
            "name": name,
            "drive_id": drive_id,
        }
        return {"id": "existing", "name": name}

    def fake_create_folder(service, parent_id, name, drive_id):
        created["created"] = {
            "parent_id": parent_id,
            "name": name,
            "drive_id": drive_id,
        }
        return {"id": "new", "name": name}

    def fake_ensure_public_permission(service, file_id):
        ensured.append(file_id)
        return {"id": file_id}

    monkeypatch.setattr(
        "drive_audit.google_client.find_child_folder", fake_find_child_folder
    )
    monkeypatch.setattr("drive_audit.google_client.create_folder", fake_create_folder)
    monkeypatch.setattr(
        "drive_audit.google_client.ensure_public_permission",
        fake_ensure_public_permission,
    )

    folder = ensure_public_subdir(
        None, parent_id="parent", subdir_name="public", drive_id="drive"
    )

    assert folder["id"] == "existing"
    assert "created" not in created
    assert ensured == ["existing"]

1	from __future__ import annotations	1✔
2
3	from pathlib import Path	1✔
4	from typing import Dict, List	1✔
5
6	from drive_audit.google_client import (	1✔
7	ensure_public_subdir,
8	list_folder_children,
9	reset_list_folder_children_cache,
10	set_list_folder_children_cache_dir,
11	)
12
13
14	class FakeRequest:	1✔
15	def __init__(self, response: Dict):	1✔
16	self.response = response	1✔
17
18	def execute(self) -> Dict:	1✔
19	return self.response	1✔
20
21
22	class FakeFilesResource:	1✔
23	def __init__(self, pages: List[Dict]):	1✔
24	self.pages = list(pages)	1✔
25	self.calls = 0	1✔
26
27	def list(self, **kwargs):	1✔
28	response = (	1✔
29	self.pages[self.calls] if self.calls < len(self.pages) else {"files": []}
30	)
31	self.calls += 1	1✔
32	return FakeRequest(response)	1✔
33
34
35	class FakeService:	1✔
36	def __init__(self, pages: List[Dict]):	1✔
37	self.files_resource = FakeFilesResource(pages)	1✔
38
39	def files(self):	1✔
40	return self.files_resource	1✔
41
42
43	def test_list_folder_children_caches_results(tmp_path: Path):	1✔
44	set_list_folder_children_cache_dir(tmp_path)	1✔
45	reset_list_folder_children_cache()	1✔
46
47	pages = [	1✔
48	{
49	"files": [
50	{
51	"id": "1",
52	"name": "first",
53	"mimeType": "text/plain",
54	"parents": ["folder"],
55	},
56	],
57	"nextPageToken": "next",
58	},
59	{
60	"files": [
61	{
62	"id": "2",
63	"name": "second",
64	"mimeType": "text/plain",
65	"parents": ["folder"],
66	}
67	]
68	},
69	]
70	service = FakeService(pages)	1✔
71
72	first_pass = list(	1✔
73	list_folder_children(
74	service,
75	folder_id="folder",
76	drive_id="drive",
77	page_size=10,
78	cache_timeout_seconds=3600,
79	)
80	)
81
82	assert [file["id"] for file in first_pass] == ["1", "2"]	1✔
83	assert service.files_resource.calls == 2	1✔
84	assert list(tmp_path.glob("*.json"))	1✔
85
86	second_pass = list(	1✔
87	list_folder_children(
88	service,
89	folder_id="folder",
90	drive_id="drive",
91	page_size=10,
92	cache_timeout_seconds=3600,
93	)
94	)
95
96	assert [file["id"] for file in second_pass] == ["1", "2"]	1✔
97	assert service.files_resource.calls == 2	1✔
98
99
100	def test_list_folder_children_skips_cache_when_disabled(tmp_path: Path):	1✔
101	set_list_folder_children_cache_dir(tmp_path)	1✔
102	reset_list_folder_children_cache()	1✔
103
104	first_pages = [	1✔
105	{
106	"files": [
107	{
108	"id": "cached",
109	"name": "cached",
110	"mimeType": "text/plain",
111	"parents": ["folder"],
112	}
113	]
114	}
115	]
116	first_pass = list(	1✔
117	list_folder_children(
118	FakeService(first_pages),
119	folder_id="folder",
120	drive_id="",
121	page_size=5,
122	cache_timeout_seconds=0,
123	)
124	)
125	second_pages = [	1✔
126	{
127	"files": [
128	{
129	"id": "cached",
130	"name": "cached",
131	"mimeType": "text/plain",
132	"parents": ["folder"],
133	}
134	]
135	}
136	]
137	second_pass = list(	1✔
138	list_folder_children(
139	FakeService(second_pages),
140	folder_id="folder",
141	drive_id="",
142	page_size=5,
143	cache_timeout_seconds=0,
144	)
145	)
146
147	assert [file["id"] for file in first_pass] == ["cached"]	1✔
148	assert [file["id"] for file in second_pass] == ["cached"]	1✔
149	assert not list(tmp_path.glob("*.json"))	1✔
150
151
152	def test_ensure_public_subdir_reuses_existing(monkeypatch):	1✔
153	created: Dict[str, Dict] = {}	1✔
154	ensured: list[str] = []	1✔
155
156	def fake_find_child_folder(service, parent_id, name, drive_id):	1✔
157	created["last_lookup"] = {	1✔
158	"parent_id": parent_id,
159	"name": name,
160	"drive_id": drive_id,
161	}
162	return {"id": "existing", "name": name}	1✔
163
164	def fake_create_folder(service, parent_id, name, drive_id):	1✔
NEW 165	created["created"] = {	×
166	"parent_id": parent_id,
167	"name": name,
168	"drive_id": drive_id,
169	}
UNCOV 170	return {"id": "new", "name": name}	×
171
172	def fake_ensure_public_permission(service, file_id):	1✔
173	ensured.append(file_id)	1✔
174	return {"id": file_id}	1✔
175
176	monkeypatch.setattr(	1✔
177	"drive_audit.google_client.find_child_folder", fake_find_child_folder
178	)
179	monkeypatch.setattr("drive_audit.google_client.create_folder", fake_create_folder)	1✔
180	monkeypatch.setattr(	1✔
181	"drive_audit.google_client.ensure_public_permission",
182	fake_ensure_public_permission,
183	)
184
185	folder = ensure_public_subdir(	1✔
186	None, parent_id="parent", subdir_name="public", drive_id="drive"
187	)
188
189	assert folder["id"] == "existing"	1✔
190	assert "created" not in created	1✔
191	assert ensured == ["existing"]	1✔

popstas / google-drive-access / 19720685265

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