19720684549

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

Build # 19720684549

Build Type

push

github

Committed by

web-flow

Commit Message

chore: add coverage workflow and caching tests (#12)

* chore: add coverage workflow and caching tests

* chore: fix test imports for package layout

Run Details

77 of 79 new or added lines in 5 files covered. (97.47%)

382 existing lines in 12 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 = self.pages[self.calls] if self.calls < len(self.pages) else {"files": []}	1✔
29	self.calls += 1	1✔
30	return FakeRequest(response)	1✔
31
32
33	class FakeService:	1✔
34	def __init__(self, pages: List[Dict]):	1✔
35	self.files_resource = FakeFilesResource(pages)	1✔
36
37	def files(self):	1✔
38	return self.files_resource	1✔
39
40
41	def test_list_folder_children_caches_results(tmp_path: Path):	1✔
42	set_list_folder_children_cache_dir(tmp_path)	1✔
43	reset_list_folder_children_cache()	1✔
44
45	pages = [	1✔
46	{
47	"files": [
48	{"id": "1", "name": "first", "mimeType": "text/plain", "parents": ["folder"]},
49	],
50	"nextPageToken": "next",
51	},
52	{"files": [{"id": "2", "name": "second", "mimeType": "text/plain", "parents": ["folder"]}]},
53	]
54	service = FakeService(pages)	1✔
55
56	first_pass = list(	1✔
57	list_folder_children(
58	service,
59	folder_id="folder",
60	drive_id="drive",
61	page_size=10,
62	cache_timeout_seconds=3600,
63	)
64	)
65
66	assert [file["id"] for file in first_pass] == ["1", "2"]	1✔
67	assert service.files_resource.calls == 2	1✔
68	assert list(tmp_path.glob("*.json"))	1✔
69
70	second_pass = list(	1✔
71	list_folder_children(
72	service,
73	folder_id="folder",
74	drive_id="drive",
75	page_size=10,
76	cache_timeout_seconds=3600,
77	)
78	)
79
80	assert [file["id"] for file in second_pass] == ["1", "2"]	1✔
81	assert service.files_resource.calls == 2	1✔
82
83
84	def test_list_folder_children_skips_cache_when_disabled(tmp_path: Path):	1✔
85	set_list_folder_children_cache_dir(tmp_path)	1✔
86	reset_list_folder_children_cache()	1✔
87
88	first_pages = [	1✔
89	{"files": [{"id": "cached", "name": "cached", "mimeType": "text/plain", "parents": ["folder"]}]}
90	]
91	first_pass = list(	1✔
92	list_folder_children(
93	FakeService(first_pages),
94	folder_id="folder",
95	drive_id="",
96	page_size=5,
97	cache_timeout_seconds=0,
98	)
99	)
100	second_pages = [	1✔
101	{"files": [{"id": "cached", "name": "cached", "mimeType": "text/plain", "parents": ["folder"]}]}
102	]
103	second_pass = list(	1✔
104	list_folder_children(
105	FakeService(second_pages),
106	folder_id="folder",
107	drive_id="",
108	page_size=5,
109	cache_timeout_seconds=0,
110	)
111	)
112
113	assert [file["id"] for file in first_pass] == ["cached"]	1✔
114	assert [file["id"] for file in second_pass] == ["cached"]	1✔
115	assert not list(tmp_path.glob("*.json"))	1✔
116
117
118	def test_ensure_public_subdir_reuses_existing(monkeypatch):	1✔
119	created: Dict[str, Dict] = {}	1✔
120	ensured: list[str] = []	1✔
121
122	def fake_find_child_folder(service, parent_id, name, drive_id):	1✔
123	created["last_lookup"] = {"parent_id": parent_id, "name": name, "drive_id": drive_id}	1✔
124	return {"id": "existing", "name": name}	1✔
125
126	def fake_create_folder(service, parent_id, name, drive_id):	1✔
NEW 127	created["created"] = {"parent_id": parent_id, "name": name, "drive_id": drive_id}	×
NEW 128	return {"id": "new", "name": name}	×
129
130	def fake_ensure_public_permission(service, file_id):	1✔
131	ensured.append(file_id)	1✔
132	return {"id": file_id}	1✔
133
134	monkeypatch.setattr("drive_audit.google_client.find_child_folder", fake_find_child_folder)	1✔
135	monkeypatch.setattr("drive_audit.google_client.create_folder", fake_create_folder)	1✔
136	monkeypatch.setattr(	1✔
137	"drive_audit.google_client.ensure_public_permission", fake_ensure_public_permission
138	)
139
140	folder = ensure_public_subdir(None, parent_id="parent", subdir_name="public", drive_id="drive")	1✔
141
142	assert folder["id"] == "existing"	1✔
143	assert "created" not in created	1✔
144	assert ensured == ["existing"]	1✔

popstas / google-drive-access / 19720684549

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