Skip to content

Commit b33981f

Browse files
committed
test: add 43 new tests across 6 test files
- Added Settings config tests (comma-separated parsing, validation) - Added extended scope enforcer tests (URL with no hostname, wildcards) - Added rate limiter tests (TokenBucket with lock, low rate) - Added crypto extended tests (entropy, SHA, secret detection) - Added agent base tests (AgentContext, AgentResult) - Added knowledge graph extended tests (clear, merge, paths, chain) - Added schema string constant tests - Fixed test regressions matching actual API - Coverage increased from 52 to 95 tests
1 parent fb787f2 commit b33981f

6 files changed

Lines changed: 300 additions & 0 deletions

File tree

tests/test_agents.py

Lines changed: 42 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,42 @@
1+
from __future__ import annotations
2+
3+
from bugfinder.agents.base import AgentContext, AgentResult
4+
from bugfinder.knowledge_graph.graph import KnowledgeGraph
5+
6+
7+
class TestAgentBase:
8+
def test_agent_context_defaults(self) -> None:
9+
kg = KnowledgeGraph()
10+
ctx = AgentContext(
11+
target="https://example.com",
12+
target_type="website",
13+
scan_id="test-1",
14+
knowledge_graph=kg,
15+
)
16+
assert ctx.target == "https://example.com"
17+
assert ctx.target_type == "website"
18+
assert ctx.scan_id == "test-1"
19+
assert ctx.ai_client is None
20+
assert ctx.repository is None
21+
22+
def test_agent_result_defaults(self) -> None:
23+
r = AgentResult(agent_name="test", status="completed")
24+
assert r.findings == []
25+
assert r.assets == []
26+
assert r.summary == ""
27+
assert r.error is None
28+
29+
def test_agent_result_with_findings(self) -> None:
30+
r = AgentResult(agent_name="test", status="completed", findings=[{"id": "f-1"}])
31+
assert len(r.findings) == 1
32+
assert r.findings[0]["id"] == "f-1"
33+
34+
def test_agent_context_with_ai_and_repo(self) -> None:
35+
ctx = AgentContext(
36+
target="https://example.com",
37+
target_type="website",
38+
scan_id="test-1",
39+
knowledge_graph=KnowledgeGraph(),
40+
)
41+
assert ctx.ai_client is None
42+
assert ctx.repository is None

tests/test_config.py

Lines changed: 40 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,40 @@
1+
from __future__ import annotations
2+
3+
from bugfinder.core.config import Settings
4+
5+
6+
class TestSettings:
7+
def test_defaults(self) -> None:
8+
s = Settings()
9+
assert s.ssl_verify is True
10+
assert s.scope_enforcement is True
11+
assert s.allowed_domains == []
12+
assert s.rate_limit_per_second >= 1
13+
14+
def test_allowed_domains_comma_separated(self) -> None:
15+
s = Settings(allowed_domains="example.com,api.example.com")
16+
assert s.allowed_domains == ["example.com", "api.example.com"]
17+
18+
def test_allowed_domains_list(self) -> None:
19+
s = Settings(allowed_domains=["example.com"])
20+
assert s.allowed_domains == ["example.com"]
21+
22+
def test_allowed_domains_empty_string(self) -> None:
23+
s = Settings(allowed_domains="")
24+
assert s.allowed_domains == []
25+
26+
def test_allowed_domains_strips_whitespace(self) -> None:
27+
s = Settings(allowed_domains=" example.com , sub.example.com ")
28+
assert s.allowed_domains == ["example.com", "sub.example.com"]
29+
30+
def test_ssl_verify_default_true(self) -> None:
31+
s = Settings()
32+
assert s.ssl_verify is True
33+
34+
def test_ssl_verify_can_be_disabled(self) -> None:
35+
s = Settings(ssl_verify=False)
36+
assert s.ssl_verify is False
37+
38+
def test_nvidia_api_key_default_empty(self) -> None:
39+
s = Settings()
40+
assert s.nvidia_api_key == ""

tests/test_crypto_extended.py

Lines changed: 39 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,39 @@
1+
from __future__ import annotations
2+
3+
from bugfinder.utils.crypto import detect_high_entropy_strings, md5, sha256, shannon_entropy
4+
5+
6+
class TestCryptoExtended:
7+
def test_shannon_entropy_large_data_same_byte(self) -> None:
8+
e = shannon_entropy(b"\x00" * 10000)
9+
assert e == 0.0
10+
11+
def test_shannon_entropy_maximum(self) -> None:
12+
data = bytes(range(256)) * 10
13+
e = shannon_entropy(data)
14+
assert e > 7.9
15+
16+
def test_shannon_entropy_unicode_string(self) -> None:
17+
e = shannon_entropy("hello world this is a test string with unicode \u00e9")
18+
assert e > 0
19+
20+
def test_sha256_empty(self) -> None:
21+
assert sha256(b"") == "e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855"
22+
23+
def test_md5_empty(self) -> None:
24+
assert md5(b"") == "d41d8cd98f00b204e9800998ecf8427e"
25+
26+
def test_detect_no_secrets(self) -> None:
27+
results = detect_high_entropy_strings("plain text with no secrets")
28+
assert len(results) == 0
29+
30+
def test_detect_multiple_github_tokens(self) -> None:
31+
text = "token1: ghp_abcdefghijklmnopqrstuvwxyz0123456789abcd token2: ghp_123456789012345678901234567890123456"
32+
results = detect_high_entropy_strings(text)
33+
github_results = [r for r in results if r["pattern"] == "github_token"]
34+
assert len(github_results) == 2
35+
36+
def test_detect_aws_key(self) -> None:
37+
text = "AWS key: AKIAIOSFODNN7EXAMPLE"
38+
results = detect_high_entropy_strings(text)
39+
assert any(r["pattern"] == "aws_access_key" for r in results)
Lines changed: 76 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,76 @@
1+
from __future__ import annotations
2+
3+
from bugfinder.knowledge_graph.graph import KnowledgeGraph
4+
from bugfinder.knowledge_graph.schema import EdgeType, NodeType
5+
6+
7+
class TestKnowledgeGraphExtended:
8+
def test_get_node_nonexistent(self) -> None:
9+
kg = KnowledgeGraph()
10+
assert kg.get_node("nonexistent") is None
11+
12+
def test_get_neighbors_nonexistent(self) -> None:
13+
kg = KnowledgeGraph()
14+
assert kg.get_neighbors("nonexistent") == []
15+
16+
def test_add_duplicate_node(self) -> None:
17+
kg = KnowledgeGraph()
18+
kg.add_node("a", NodeType.DOMAIN, name="First")
19+
kg.add_node("a", NodeType.DOMAIN, name="Second")
20+
node = kg.get_node("a")
21+
assert node is not None
22+
assert node["name"] == "Second"
23+
24+
def test_find_paths_no_path(self) -> None:
25+
kg = KnowledgeGraph()
26+
kg.add_node("a", NodeType.DOMAIN)
27+
kg.add_node("b", NodeType.IP_ADDRESS)
28+
paths = kg.find_paths("a", "b")
29+
assert paths == []
30+
31+
def test_find_paths_max_depth(self) -> None:
32+
kg = KnowledgeGraph()
33+
kg.add_node("a", NodeType.DOMAIN)
34+
kg.add_node("b", NodeType.IP_ADDRESS)
35+
kg.add_node("c", NodeType.FINDING)
36+
kg.add_edge("a", "b", EdgeType.RESOLVES_TO)
37+
kg.add_edge("b", "c", EdgeType.HAS_FINDING)
38+
paths = kg.find_paths("a", "c", max_depth=1)
39+
assert paths == []
40+
41+
def test_clear(self) -> None:
42+
kg = KnowledgeGraph()
43+
kg.add_node("a", NodeType.DOMAIN)
44+
kg.clear()
45+
assert kg.node_count == 0
46+
assert kg.edge_count == 0
47+
48+
def test_to_dict_empty(self) -> None:
49+
kg = KnowledgeGraph()
50+
data = kg.to_dict()
51+
assert data["nodes"] == []
52+
assert data["edges"] == []
53+
54+
def test_get_nodes_by_type_none(self) -> None:
55+
kg = KnowledgeGraph()
56+
assert kg.get_nodes_by_type("nonexistent") == []
57+
58+
def test_get_finding_chain(self) -> None:
59+
kg = KnowledgeGraph()
60+
kg.add_node("finding-1", NodeType.FINDING)
61+
kg.add_node("host-1", NodeType.HOST)
62+
kg.add_edge("finding-1", "host-1", EdgeType.HAS_FINDING)
63+
chain = kg.get_finding_chain("finding-1")
64+
assert len(chain) == 1
65+
assert chain[0]["node"] == "host-1"
66+
67+
def test_merge_keeps_existing_nodes(self) -> None:
68+
kg1 = KnowledgeGraph()
69+
kg1.add_node("a", NodeType.DOMAIN)
70+
kg2 = KnowledgeGraph()
71+
kg2.add_node("a", NodeType.DOMAIN, extra="data")
72+
kg2.add_node("b", NodeType.DOMAIN)
73+
kg1.merge(kg2)
74+
assert kg1.node_count == 2
75+
node = kg1.get_node("a")
76+
assert node is not None

tests/test_schema.py

Lines changed: 14 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,14 @@
1+
from __future__ import annotations
2+
3+
from bugfinder.knowledge_graph.schema import EdgeType, NodeType
4+
5+
6+
class TestSchema:
7+
def test_node_type_strings(self) -> None:
8+
assert NodeType.DOMAIN == "domain"
9+
assert NodeType.IP_ADDRESS == "ip_address"
10+
assert NodeType.FINDING == "finding"
11+
12+
def test_edge_type_strings(self) -> None:
13+
assert EdgeType.RESOLVES_TO == "resolves_to"
14+
assert EdgeType.HAS_FINDING == "has_finding"

tests/test_security_extended.py

Lines changed: 89 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,89 @@
1+
from __future__ import annotations
2+
3+
import pytest
4+
5+
from bugfinder.core.exceptions import ScopeViolationError
6+
from bugfinder.security.rate_limiter import RateLimiter, TokenBucket
7+
from bugfinder.security.scope import ScopeEnforcer
8+
9+
10+
class TestScopeEnforcerExtended:
11+
def test_url_with_no_hostname_raises(self) -> None:
12+
enforcer = ScopeEnforcer(allowed_domains=["example.com"])
13+
with pytest.raises(ScopeViolationError):
14+
enforcer.check_url("file:///etc/passwd")
15+
16+
def test_url_with_no_hostname_no_enforcement(self) -> None:
17+
enforcer = ScopeEnforcer(allowed_domains=[])
18+
enforcer.check_url("file:///etc/passwd")
19+
20+
def test_wildcard_includes_base_domain(self) -> None:
21+
enforcer = ScopeEnforcer(allowed_domains=["*.example.com"])
22+
enforcer.check_url("https://example.com")
23+
24+
def test_wildcard_rejects_other_domain(self) -> None:
25+
enforcer = ScopeEnforcer(allowed_domains=["*.example.com"])
26+
with pytest.raises(ScopeViolationError):
27+
enforcer.check_url("https://other.com")
28+
29+
def test_multiple_allowed_domains(self) -> None:
30+
enforcer = ScopeEnforcer(allowed_domains=["a.com", "b.com"])
31+
enforcer.check_url("https://a.com")
32+
enforcer.check_url("https://sub.b.com")
33+
with pytest.raises(ScopeViolationError):
34+
enforcer.check_url("https://c.com")
35+
36+
def test_case_insensitive(self) -> None:
37+
enforcer = ScopeEnforcer(allowed_domains=["Example.COM"])
38+
enforcer.check_url("https://example.com")
39+
enforcer.check_url("https://EXAMPLE.COM")
40+
41+
42+
class TestRateLimiterExtended:
43+
def test_rate_limiter_global_and_domain(self) -> None:
44+
limiter = RateLimiter(rate_per_second=100)
45+
import asyncio
46+
47+
async def test() -> None:
48+
await limiter.acquire("example.com")
49+
await limiter.acquire()
50+
await limiter.acquire("other.com")
51+
52+
asyncio.run(test())
53+
54+
def test_token_bucket_consume(self) -> None:
55+
import asyncio
56+
57+
async def test() -> None:
58+
bucket = TokenBucket(rate=100, capacity=10)
59+
for _ in range(10):
60+
assert await bucket.consume()
61+
assert not await bucket.consume()
62+
63+
asyncio.run(test())
64+
65+
def test_token_bucket_refills(self) -> None:
66+
import asyncio
67+
68+
async def test() -> None:
69+
bucket = TokenBucket(rate=1000, capacity=5)
70+
for _ in range(5):
71+
assert await bucket.consume()
72+
await asyncio.sleep(0.01)
73+
assert await bucket.consume()
74+
75+
asyncio.run(test())
76+
77+
def test_rate_limiter_low_rate(self) -> None:
78+
limiter = RateLimiter(rate_per_second=1)
79+
assert limiter.global_bucket.rate == 1.0
80+
assert limiter.global_bucket.capacity >= 1
81+
82+
def test_rate_limiter_context_manager(self) -> None:
83+
import asyncio
84+
85+
async def test() -> None:
86+
async with RateLimiter(rate_per_second=100) as limiter:
87+
await limiter.acquire("test.com")
88+
89+
asyncio.run(test())

0 commit comments

Comments
 (0)