Skip to content

Commit 9bb2e2e

Browse files
committed
tests(async_svn): Add AsyncSvnSync sync tests
why: Verify async SVN repository synchronization what: - Add TestAsyncSvnSync for init, repr, auth tests - Add TestAsyncSvnSyncObtain for checkout operations - Add TestAsyncSvnSyncUpdateRepo and get_revision tests
1 parent 5060613 commit 9bb2e2e

1 file changed

Lines changed: 248 additions & 0 deletions

File tree

tests/sync/_async/test_svn.py

Lines changed: 248 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,248 @@
1+
"""Tests for libvcs.sync._async.svn."""
2+
3+
from __future__ import annotations
4+
5+
import shutil
6+
from pathlib import Path
7+
8+
import pytest
9+
10+
from libvcs.pytest_plugin import CreateRepoPytestFixtureFn
11+
from libvcs.sync._async.svn import AsyncSvnSync
12+
13+
if not shutil.which("svn"):
14+
pytestmark = pytest.mark.skip(reason="svn is not available")
15+
16+
17+
class TestAsyncSvnSyncGetSvnUrlRev:
18+
"""Tests for AsyncSvnSync._get_svn_url_rev()."""
19+
20+
@pytest.mark.asyncio
21+
async def test_get_svn_url_rev_from_working_copy(
22+
self,
23+
tmp_path: Path,
24+
create_svn_remote_repo: CreateRepoPytestFixtureFn,
25+
) -> None:
26+
"""Test _get_svn_url_rev from actual working copy."""
27+
remote_repo = create_svn_remote_repo()
28+
repo_path = tmp_path / "url_rev_repo"
29+
30+
repo = AsyncSvnSync(
31+
url=f"file://{remote_repo}",
32+
path=repo_path,
33+
)
34+
await repo.obtain()
35+
36+
# Get URL and revision from the working copy
37+
url, rev = await repo._get_svn_url_rev(str(repo_path))
38+
39+
# URL should match the remote
40+
assert url is not None
41+
assert str(remote_repo) in url
42+
# Revision should be 0 for empty repo
43+
assert rev == 0
44+
45+
@pytest.mark.asyncio
46+
async def test_get_svn_url_rev_nonexistent_location(
47+
self,
48+
tmp_path: Path,
49+
create_svn_remote_repo: CreateRepoPytestFixtureFn,
50+
) -> None:
51+
"""Test _get_svn_url_rev with non-existent location."""
52+
remote_repo = create_svn_remote_repo()
53+
repo_path = tmp_path / "url_rev_repo"
54+
55+
repo = AsyncSvnSync(
56+
url=f"file://{remote_repo}",
57+
path=repo_path,
58+
)
59+
await repo.obtain()
60+
61+
# Get URL and revision from non-existent path
62+
url, rev = await repo._get_svn_url_rev(str(tmp_path / "nonexistent"))
63+
64+
# Should return None, 0 for non-existent path
65+
assert url is None
66+
assert rev == 0
67+
68+
@pytest.mark.asyncio
69+
async def test_get_svn_url_rev_xml_format(
70+
self,
71+
tmp_path: Path,
72+
create_svn_remote_repo: CreateRepoPytestFixtureFn,
73+
) -> None:
74+
"""Test _get_svn_url_rev with XML format entries file."""
75+
remote_repo = create_svn_remote_repo()
76+
repo_path = tmp_path / "xml_repo"
77+
78+
repo = AsyncSvnSync(
79+
url=f"file://{remote_repo}",
80+
path=repo_path,
81+
)
82+
await repo.obtain()
83+
84+
# Create a mock XML entries file
85+
entries_path = repo_path / ".svn" / "entries"
86+
xml_data = f"""<?xml version="1.0" encoding="utf-8"?>
87+
<wc-entries xmlns="svn:">
88+
<entry name="" url="file://{remote_repo}" committed-rev="42"/>
89+
<entry name="file.txt" committed-rev="10"/>
90+
</wc-entries>"""
91+
entries_path.write_text(xml_data)
92+
93+
# Parse the mock entries file
94+
url, rev = await repo._get_svn_url_rev(str(repo_path))
95+
96+
# Should parse URL and max revision from XML
97+
assert url is not None
98+
assert str(remote_repo) in url
99+
assert rev == 42 # max of 42 and 10
100+
101+
102+
class TestAsyncSvnSync:
103+
"""Tests for AsyncSvnSync class."""
104+
105+
def test_init(self, tmp_path: Path) -> None:
106+
"""Test AsyncSvnSync initialization."""
107+
repo = AsyncSvnSync(
108+
url="file:///path/to/repo",
109+
path=tmp_path / "repo",
110+
)
111+
assert repo.url == "file:///path/to/repo"
112+
assert repo.path == tmp_path / "repo"
113+
114+
def test_repr(self, tmp_path: Path) -> None:
115+
"""Test AsyncSvnSync repr."""
116+
repo = AsyncSvnSync(
117+
url="file:///path/to/repo",
118+
path=tmp_path / "myrepo",
119+
)
120+
assert "AsyncSvnSync" in repr(repo)
121+
assert "myrepo" in repr(repo)
122+
123+
def test_init_with_auth(self, tmp_path: Path) -> None:
124+
"""Test AsyncSvnSync initialization with auth credentials."""
125+
repo = AsyncSvnSync(
126+
url="svn://svn.example.com/repo",
127+
path=tmp_path / "repo",
128+
username="user",
129+
password="pass",
130+
svn_trust_cert=True,
131+
)
132+
assert repo.username == "user"
133+
assert repo.password == "pass"
134+
assert repo.svn_trust_cert is True
135+
136+
137+
class TestAsyncSvnSyncObtain:
138+
"""Tests for AsyncSvnSync.obtain()."""
139+
140+
@pytest.mark.asyncio
141+
async def test_obtain_basic(
142+
self,
143+
tmp_path: Path,
144+
create_svn_remote_repo: CreateRepoPytestFixtureFn,
145+
) -> None:
146+
"""Test basic obtain operation."""
147+
remote_repo = create_svn_remote_repo()
148+
repo_path = tmp_path / "obtained_repo"
149+
150+
repo = AsyncSvnSync(
151+
url=f"file://{remote_repo}",
152+
path=repo_path,
153+
)
154+
await repo.obtain()
155+
156+
assert repo_path.exists()
157+
assert (repo_path / ".svn").exists()
158+
159+
160+
class TestAsyncSvnSyncUpdateRepo:
161+
"""Tests for AsyncSvnSync.update_repo()."""
162+
163+
@pytest.mark.asyncio
164+
async def test_update_repo_basic(
165+
self,
166+
tmp_path: Path,
167+
create_svn_remote_repo: CreateRepoPytestFixtureFn,
168+
) -> None:
169+
"""Test basic update_repo operation."""
170+
remote_repo = create_svn_remote_repo()
171+
repo_path = tmp_path / "update_repo"
172+
173+
repo = AsyncSvnSync(
174+
url=f"file://{remote_repo}",
175+
path=repo_path,
176+
)
177+
# First obtain
178+
await repo.obtain()
179+
180+
# Then update
181+
await repo.update_repo()
182+
183+
assert repo_path.exists()
184+
185+
@pytest.mark.asyncio
186+
async def test_update_repo_obtains_if_missing(
187+
self,
188+
tmp_path: Path,
189+
create_svn_remote_repo: CreateRepoPytestFixtureFn,
190+
) -> None:
191+
"""Test update_repo checks out if repo doesn't exist."""
192+
remote_repo = create_svn_remote_repo()
193+
repo_path = tmp_path / "new_repo"
194+
195+
repo = AsyncSvnSync(
196+
url=f"file://{remote_repo}",
197+
path=repo_path,
198+
)
199+
# Just update_repo without obtain first
200+
await repo.update_repo()
201+
202+
# Should have checked out
203+
assert repo_path.exists()
204+
assert (repo_path / ".svn").exists()
205+
206+
207+
class TestAsyncSvnSyncGetRevision:
208+
"""Tests for AsyncSvnSync.get_revision()."""
209+
210+
@pytest.mark.asyncio
211+
async def test_get_revision(
212+
self,
213+
tmp_path: Path,
214+
create_svn_remote_repo: CreateRepoPytestFixtureFn,
215+
) -> None:
216+
"""Test get_revision returns current revision."""
217+
remote_repo = create_svn_remote_repo()
218+
repo_path = tmp_path / "rev_repo"
219+
220+
repo = AsyncSvnSync(
221+
url=f"file://{remote_repo}",
222+
path=repo_path,
223+
)
224+
await repo.obtain()
225+
226+
revision = await repo.get_revision()
227+
# SVN revisions start at 0 for empty repos
228+
assert revision == 0
229+
230+
@pytest.mark.asyncio
231+
async def test_get_revision_file(
232+
self,
233+
tmp_path: Path,
234+
create_svn_remote_repo: CreateRepoPytestFixtureFn,
235+
) -> None:
236+
"""Test get_revision_file returns file revision."""
237+
remote_repo = create_svn_remote_repo()
238+
repo_path = tmp_path / "rev_file_repo"
239+
240+
repo = AsyncSvnSync(
241+
url=f"file://{remote_repo}",
242+
path=repo_path,
243+
)
244+
await repo.obtain()
245+
246+
revision = await repo.get_revision_file("./")
247+
# SVN revisions start at 0 for empty repos
248+
assert revision == 0

0 commit comments

Comments
 (0)