| """ |
| Tests for TokenSupplyAnalyzer |
| """ |
|
|
| import unittest |
| from datetime import UTC, datetime |
|
|
| from app.token_supply_analyzer import ( |
| LockStatus, |
| SupplyRisk, |
| TokenSupplyProfile, |
| _analyze_concentration, |
| _analyze_liquidity, |
| _analyze_supply_mechanics, |
| _classify_risk, |
| _compute_overall_risk, |
| _compute_pair_age_days, |
| _detect_chain, |
| _normalize_address, |
| ) |
|
|
|
|
| class TestSupplyScoring(unittest.TestCase): |
| """Test the supply mechanics analysis engine.""" |
|
|
| def test_healthy_supply(self) -> None: |
| """A token with normal supply metrics should score low risk.""" |
| profile = TokenSupplyProfile(address="0xtest123") |
| profile.total_supply = 1_000_000_000 |
| profile.circulating_supply = 800_000_000 |
| profile.max_supply = 1_000_000_000 |
| profile.burned_supply = 50_000_000 |
| profile.has_mint_function = False |
| profile.is_deflationary = True |
|
|
| _analyze_supply_mechanics(profile) |
| self.assertLess( |
| profile.supply_risk_score, |
| 30, |
| f"Healthy supply should score low, got {profile.supply_risk_score}", |
| ) |
|
|
| def test_unlimited_mint_supply(self) -> None: |
| """A token with no max supply and mint function should score high.""" |
| profile = TokenSupplyProfile(address="0xevil123") |
| profile.total_supply = 1_000_000 |
| profile.circulating_supply = 100_000 |
| profile.max_supply = 0 |
| profile.has_mint_function = True |
| profile.mint_function_type = "unlimited" |
|
|
| _analyze_supply_mechanics(profile) |
| self.assertGreater( |
| profile.supply_risk_score, |
| 50, |
| f"Unlimited mint should score high, got {profile.supply_risk_score}", |
| ) |
| self.assertIn("unlimited_mint_function", str(profile.patterns_detected)) |
|
|
| def test_supply_exceeds_max(self) -> None: |
| """A token where total > max supply should be critical.""" |
| profile = TokenSupplyProfile(address="0xoverflow") |
| profile.total_supply = 2_000_000_000 |
| profile.circulating_supply = 1_500_000_000 |
| profile.max_supply = 1_000_000_000 |
|
|
| _analyze_supply_mechanics(profile) |
| self.assertGreaterEqual( |
| profile.supply_risk_score, |
| 40, |
| f"Supply overflow should score high, got {profile.supply_risk_score}", |
| ) |
| self.assertIn("supply_exceeds_max", str(profile.patterns_detected)) |
|
|
| def test_dilution_risk(self) -> None: |
| """A token with <30% circ/total ratio should flag dilution.""" |
| profile = TokenSupplyProfile(address="0xdilute") |
| profile.total_supply = 10_000_000 |
| profile.circulating_supply = 2_000_000 |
|
|
| _analyze_supply_mechanics(profile) |
| self.assertIn("high_dilution_risk", str(profile.patterns_detected)) |
|
|
| def test_large_mint_headroom(self) -> None: |
| """A token with >50% of max supply remaining should flag.""" |
| profile = TokenSupplyProfile(address="0xmintroom") |
| profile.total_supply = 100_000 |
| profile.circulating_supply = 80_000 |
| profile.max_supply = 1_000_000 |
| profile.has_mint_function = True |
|
|
| _analyze_supply_mechanics(profile) |
| self.assertIn("large_mint_headroom", str(profile.patterns_detected)) |
|
|
|
|
| class TestConcentrationScoring(unittest.TestCase): |
| """Test holder concentration analysis.""" |
|
|
| def test_extreme_concentration(self) -> None: |
| """Top 10 holding >99% should be critical.""" |
| profile = TokenSupplyProfile(address="0xconcentrated") |
| profile.top_10_holder_pct = 99.5 |
| profile.top_50_holder_pct = 99.9 |
|
|
| _analyze_concentration(profile) |
| self.assertIn("extreme_concentration:top_10_holds_>99%", str(profile.patterns_detected)) |
| self.assertIn(profile.concentration_risk, (SupplyRisk.HIGH, SupplyRisk.CRITICAL)) |
|
|
| def test_deployer_holds_majority(self) -> None: |
| """Deployer holding >50% should be critical.""" |
| profile = TokenSupplyProfile(address="0xdeployer_owns") |
| profile.deployer_hold_pct = 75.0 |
| profile.top_10_holder_pct = 85.0 |
|
|
| _analyze_concentration(profile) |
| self.assertIn("deployer_holds_majority", str(profile.patterns_detected)) |
|
|
| def test_no_concentration(self) -> None: |
| """No significant concentration should score low.""" |
| profile = TokenSupplyProfile(address="0xfair_dist") |
| profile.deployer_hold_pct = 2.0 |
| profile.top_10_holder_pct = 15.0 |
| profile.top_50_holder_pct = 30.0 |
|
|
| _analyze_concentration(profile) |
| self.assertLess( |
| profile.concentration_score, |
| 20, |
| f"Fair distribution should score low, got {profile.concentration_score}", |
| ) |
|
|
|
|
| class TestLiquidityScoring(unittest.TestCase): |
| """Test liquidity analysis.""" |
|
|
| def test_liquidity_removed(self) -> None: |
| """Removed liquidity should be critical.""" |
| profile = TokenSupplyProfile(address="0xrugpull") |
| profile.lock_status = LockStatus.REMOVED |
|
|
| _analyze_liquidity(profile) |
| self.assertIn("liquidity_removed", str(profile.patterns_detected)) |
| self.assertGreater(profile.liquidity_score, 40) |
|
|
| def test_liquidity_locked(self) -> None: |
| """Locked liquidity should reduce risk.""" |
| profile = TokenSupplyProfile(address="0xsafe") |
| profile.lock_status = LockStatus.LOCKED |
| profile.liquidity_lock_expiry = "2027-01-01" |
| profile.pair_age_days = 60 |
| profile.pair_liquidity_usd = 200_000 |
|
|
| _analyze_liquidity(profile) |
| self.assertIn("liquidity_locked", str(profile.patterns_detected)) |
| self.assertLess(profile.liquidity_score, 20) |
|
|
| def test_very_new_pair(self) -> None: |
| """Very new pair should be flagged.""" |
| profile = TokenSupplyProfile(address="0xnew_token") |
| profile.pair_age_days = 0.5 |
| profile.pair_liquidity_usd = 500 |
|
|
| _analyze_liquidity(profile) |
| self.assertIn("very_new_pair", str(profile.patterns_detected)) |
| self.assertIn("very_low_liquidity", str(profile.patterns_detected)) |
|
|
|
|
| class TestOverallRisk(unittest.TestCase): |
| """Test overall risk computation.""" |
|
|
| def test_critical_token(self) -> None: |
| """A token with multiple red flags should be critical.""" |
| profile = TokenSupplyProfile(address="0xscam_token") |
| profile.has_mint_function = True |
| profile.mint_function_type = "unlimited" |
| profile.has_dynamic_tax = True |
| profile.has_blacklist = True |
| profile.buy_tax_pct = 15.0 |
| profile.sell_tax_pct = 25.0 |
| profile.lock_status = LockStatus.REMOVED |
| profile.top_10_holder_pct = 99.0 |
| profile.is_renounced = False |
| profile.has_proxy_admin = True |
|
|
| _analyze_supply_mechanics(profile) |
| _analyze_concentration(profile) |
| _analyze_liquidity(profile) |
| _compute_overall_risk(profile) |
| self.assertIn( |
| profile.overall_risk, |
| (SupplyRisk.HIGH, SupplyRisk.CRITICAL), |
| f"Scam token should be high/critical, got {profile.overall_risk}", |
| ) |
|
|
| def test_safe_token(self) -> None: |
| """A token with good practices should be safe/low.""" |
| profile = TokenSupplyProfile(address="0xgood_token") |
| profile.total_supply = 1_000_000 |
| profile.circulating_supply = 950_000 |
| profile.max_supply = 1_000_000 |
| profile.burned_supply = 50_000 |
| profile.has_mint_function = False |
| profile.is_deflationary = True |
| profile.buy_tax_pct = 1.0 |
| profile.sell_tax_pct = 1.0 |
| profile.lock_status = LockStatus.LOCKED |
| profile.is_renounced = True |
| profile.top_10_holder_pct = 20.0 |
| profile.top_50_holder_pct = 35.0 |
| profile.deployer_hold_pct = 3.0 |
| profile.pair_age_days = 120 |
| profile.pair_liquidity_usd = 500_000 |
|
|
| _analyze_supply_mechanics(profile) |
| _analyze_concentration(profile) |
| _analyze_liquidity(profile) |
| _compute_overall_risk(profile) |
| self.assertIn( |
| profile.overall_risk, |
| (SupplyRisk.SAFE, SupplyRisk.LOW), |
| f"Safe token should be safe/low, got {profile.overall_risk}", |
| ) |
|
|
|
|
| class TestUtilities(unittest.TestCase): |
| """Test utility functions.""" |
|
|
| def test_detect_chain(self) -> None: |
| """Chain detection should work for common formats.""" |
| self.assertEqual(_detect_chain("0xabcd1234"), "ethereum") |
| self.assertEqual( |
| _detect_chain("AbCdEf1234567890AbCdEf1234567890AbCdEf1234567890AbCdEf1234567890"), |
| "solana", |
| ) |
| self.assertEqual(_detect_chain("unknown"), "unknown") |
|
|
| def test_normalize_address(self) -> None: |
| """Address normalization should lowercase and strip.""" |
| self.assertEqual(_normalize_address("0xABC123"), "0xabc123") |
| self.assertEqual(_normalize_address(" 0xABC "), "0xabc") |
|
|
| def test_risk_classification(self) -> None: |
| """Risk classification boundaries.""" |
| self.assertEqual(_classify_risk(0), SupplyRisk.SAFE) |
| self.assertEqual(_classify_risk(10), SupplyRisk.SAFE) |
| self.assertEqual(_classify_risk(11), SupplyRisk.LOW) |
| self.assertEqual(_classify_risk(25), SupplyRisk.LOW) |
| self.assertEqual(_classify_risk(26), SupplyRisk.MEDIUM) |
| self.assertEqual(_classify_risk(50), SupplyRisk.MEDIUM) |
| self.assertEqual(_classify_risk(51), SupplyRisk.HIGH) |
| self.assertEqual(_classify_risk(75), SupplyRisk.HIGH) |
| self.assertEqual(_classify_risk(76), SupplyRisk.CRITICAL) |
|
|
| def test_pair_age_computation(self) -> None: |
| """Pair age from Unix ms timestamp.""" |
| now = datetime.now(tz=UTC) |
| one_day_ago = int((now.timestamp() - 86400) * 1000) |
| age = _compute_pair_age_days(one_day_ago) |
| self.assertAlmostEqual(age, 1.0, delta=0.1) |
|
|
| |
| self.assertEqual(_compute_pair_age_days(0), 0.0) |
|
|
| |
| future = int((now.timestamp() + 86400) * 1000) |
| self.assertEqual(_compute_pair_age_days(future), 0.0) |
|
|
|
|
| class TestOwnershipAndTax(unittest.TestCase): |
| """Test ownership and tax analysis in overall risk.""" |
|
|
| def test_proxy_admin_risk(self) -> None: |
| """Proxy admin should flag upgrade risk.""" |
| profile = TokenSupplyProfile(address="0xproxy") |
| profile.has_proxy_admin = True |
| profile.is_renounced = False |
|
|
| _compute_overall_risk(profile) |
| self.assertIn("proxy_admin:contract_can_be_upgraded", str(profile.patterns_detected)) |
|
|
| def test_multisig_benefit(self) -> None: |
| """Multi-sig ownership should reduce risk.""" |
| profile = TokenSupplyProfile(address="0xmultisig") |
| profile.has_multisig = True |
| profile.is_renounced = False |
|
|
| _compute_overall_risk(profile) |
| self.assertIn("multisig_owner", str(profile.patterns_detected)) |
|
|
| def test_dynamic_tax_detection(self) -> None: |
| """Dynamic tax should be flagged.""" |
| profile = TokenSupplyProfile(address="0xtaxy") |
| profile.has_dynamic_tax = True |
| profile.has_blacklist = True |
|
|
| _compute_overall_risk(profile) |
| self.assertIn("dynamic_tax:tax_can_change", str(profile.patterns_detected)) |
| self.assertIn("blacklist_function", str(profile.patterns_detected)) |
|
|
| def test_high_sell_tax(self) -> None: |
| """Very high sell tax should flag honeypot risk.""" |
| profile = TokenSupplyProfile(address="0xhoneypot") |
| profile.sell_tax_pct = 99.0 |
|
|
| _compute_overall_risk(profile) |
| self.assertIn("very_high_sell_tax:99.0%", str(profile.patterns_detected)) |
|
|
|
|
| if __name__ == "__main__": |
| unittest.main() |
|
|
