chore: Rule framework now use RuleExplorer a MightstoneHttpClient

Removed dependency to Requests, and is async ready.

pytest.ini

@@ -1,3 +1,4 @@
 [pytest]
 markers =
     skip_remote_api:Skips remote api calls on CI
+    skip_remote_api_2:Skips remote api calls on CI (same thing but i don’t know how to relative import)

src/mightstone/rule/cr.py

@@ -1,21 +1,14 @@
 import asyncio
-import logging
 import re
 from collections import defaultdict
 from datetime import date, datetime, timedelta
 from io import StringIO
 from itertools import takewhile
 from typing import Dict, List, Mapping, TextIO
 
-import httpx
-import requests
-
+from mightstone import logger
 from mightstone.core import MightstoneModel
-
-# TODO: get rid of requests dependency, use async client
-
-
-logger = logging.getLogger(__name__)
+from mightstone.services import MightstoneHttpClient
 
 
 class RuleRef(str):
@@ -267,7 +260,7 @@ def index(self):
 
 
 class ComprehensiveRules(MightstoneModel):
-    effective: date = None
+    effective: Effectiveness = None
     ruleset: Ruleset = Ruleset()
     glossary: Glossary = Glossary()
 
@@ -278,7 +271,7 @@ def search(self, string):
         return found
 
     @classmethod
-    def from_text(cls, buffer: TextIO):
+    def parse(cls, buffer: TextIO):
         cr = ComprehensiveRules()
         in_glossary = False
         in_credits = False
@@ -322,79 +315,6 @@ def from_text(cls, buffer: TextIO):
 
         return cr
 
-    @classmethod
-    def from_file(cls, path):
-        with open(path, "r") as f:
-            return cls.from_text(f)
-
-    @classmethod
-    def from_url(cls, url: str):
-        with requests.get(url) as f:
-            f.raise_for_status()
-            try:
-                content = StringIO(f.content.decode("UTF-8"))
-            except UnicodeDecodeError:
-                content = StringIO(f.content.decode("iso-8859-1"))
-
-            return cls.from_text(content)
-
-    @classmethod
-    def from_latest(cls):
-        latest = cls.latest()
-        return cls.from_url(latest)
-
-    @classmethod
-    def latest(cls):
-        pattern = re.compile(r"https://media.wizards.com/.*/MagicComp.+\.txt")
-        with requests.get("https://magic.wizards.com/en/rules") as f:
-            f.raise_for_status()
-            res = pattern.search(f.text)
-            if res:
-                return res.group(0).replace(" ", "%20")
-            raise RuntimeError("Unable to find URL of the last comprehensive rules")
-
-    @classmethod
-    async def explore(cls, f: date, t: date = None, concurrency=3):
-        """
-        AFAIK Wizards don’t support an historic index of previous rules.
-        This method will force try every possible rule using the current format:
-
-        https://media.wizards.com/YYYY/downloads/MagicComp%20Rules%20{YYYY}{MM}{DD}.txt
-        :param f: The min date to scan
-        :param t: The max date to scan (defaults to today)
-        :param concurrency: The max number of concurrent HTTP requests
-        :return: A list of existing rules url
-        """
-        if not t:
-            t = date.today()
-
-        urls = []
-        for n in range(int((t - f).days)):
-            d = f + timedelta(n)
-            urls.append(d.strftime("/%Y/downloads/MagicComp%%20Rules%%20%Y%m%d.txt"))
-            urls.append(d.strftime("/%Y/downloads/MagicCompRules%%20%Y%m%d.txt"))
-        map(lambda x: f"https://media.wizards.com{x}", urls)
-
-        found = []
-        sem = asyncio.Semaphore(concurrency)
-
-        async def test_url(session, url):
-            async with sem:
-                logger.debug("GET %s", url)
-                resp = session.get(url)
-                if resp.status == 200:
-                    logger.info("Found %s", url)
-                    found.append(url)
-
-        async with httpx.Client() as session:
-            tasks = []
-            for url in urls:
-                task = asyncio.ensure_future(test_url(session, url))
-                tasks.append(task)
-            await asyncio.gather(*tasks, return_exceptions=True)
-
-        return found
-
     def diff(self, cr: "ComprehensiveRules"):
         """
         Compare two comprehensive rule set for change in rules and terms
@@ -465,3 +385,73 @@ def diff(self, cr: "ComprehensiveRules"):
                 }
 
         return diff
+
+
+class RuleExplorer(MightstoneHttpClient):
+    base_url = "https://media.wizards.com"
+
+    async def open(self, path: str = None) -> ComprehensiveRules:
+        if not path:
+            path = await self.latest()
+
+        if path.startswith("http"):
+            f = await self.client.get(path)
+            f.raise_for_status()
+            try:
+                content = StringIO(f.content.decode("UTF-8"))
+            except UnicodeDecodeError:
+                content = StringIO(f.content.decode("iso-8859-1"))
+
+            return ComprehensiveRules.parse(content)
+
+        with open(path, "r") as f:
+            return ComprehensiveRules.parse(f)
+
+    async def latest(self) -> str:
+        pattern = re.compile(re.escape(self.base_url) + r"/.*/MagicComp.+\.txt")
+
+        f = await self.client.get("https://magic.wizards.com/en/rules")
+        f.raise_for_status()
+        res = pattern.search(f.text)
+        if res:
+            return res.group(0).replace(" ", "%20")
+        raise RuntimeError("Unable to find URL of the last comprehensive rules")
+
+    async def explore(self, f: date, t: date = None, concurrency=3) -> List[str]:
+        """
+        AFAIK Wizards don’t support an historic index of previous rules.
+        This method will force try every possible rule using the current format:
+
+        https://media.wizards.com/YYYY/downloads/MagicComp%20Rules%20{YYYY}{MM}{DD}.txt
+        :param f: The min date to scan
+        :param t: The max date to scan (defaults to today)
+        :param concurrency: The max number of concurrent HTTP requests
+        :return: A list of existing rules url
+        """
+        if not t:
+            t = date.today()
+
+        urls = []
+        for n in range(int((t - f).days)):
+            d = f + timedelta(n)
+            urls.append(d.strftime("/%Y/downloads/MagicComp%%20Rules%%20%Y%m%d.txt"))
+            urls.append(d.strftime("/%Y/downloads/MagicCompRules%%20%Y%m%d.txt"))
+
+        found = []
+        sem = asyncio.Semaphore(concurrency)
+
+        async def test_url(url: str):
+            async with sem:
+                logger.debug("GET %s", url)
+                resp = await self.client.get(url)
+                if resp.is_success:
+                    logger.info("Found %s", url)
+                    found.append(url)
+
+        tasks = []
+        for url in urls:
+            task = asyncio.ensure_future(test_url(url))
+            tasks.append(task)
+        await asyncio.gather(*tasks, return_exceptions=True)
+
+        return found

tests/mightstone/rule/__init__.py

@@ -0,0 +1,13 @@
+import os
+
+import pytest
+
+
+@pytest.fixture(autouse=True)
+def skip_remote_api_2():
+    if "CI" in os.environ:
+        pytest.skip("Remote API tests are disable in CI")
+    elif os.environ.get("REMOTE_API_TEST", "1") != "1":
+        pytest.skip("Remote API tests are disable by explicit REMOTE_API_TEST != 1")
+    else:
+        yield

tests/mightstone/rule/test_cr.py

@@ -3,18 +3,23 @@
 import unittest
 from io import StringIO
 
+import pytest as pytest
+
 from mightstone.rule.cr import (
     ComprehensiveRules,
     Effectiveness,
     Example,
     Glossary,
     Rule,
+    RuleExplorer,
     RuleRef,
     Ruleset,
     RuleText,
     SectionRef,
 )
 
+from . import skip_remote_api_2  # noqa: F401
+
 
 class TestExample(unittest.TestCase):
     def test_example(self):
@@ -432,7 +437,7 @@ def setUp(self) -> None:
         )
 
     def test_ruleset_access(self):
-        cr = ComprehensiveRules.from_text(self.buffer)
+        cr = ComprehensiveRules.parse(self.buffer)
 
         self.assertEqual(cr.ruleset["200"].text, "In hac habitasse platea dictumst.")
         self.assertEqual(
@@ -446,13 +451,13 @@ def test_ruleset_access(self):
         )
 
     def test_ruleset_key_error(self):
-        cr = ComprehensiveRules.from_text(self.buffer)
+        cr = ComprehensiveRules.parse(self.buffer)
 
         with self.assertRaises(KeyError):
             print(cr.ruleset["not_found"])
 
     def test_ruleset_reference(self):
-        cr = ComprehensiveRules.from_text(self.buffer)
+        cr = ComprehensiveRules.parse(self.buffer)
 
         self.assertEqual(len(cr.ruleset["100.1"].text.refs), 1)
         self.assertEqual(cr.ruleset["100.1"].text.refs, [RuleRef("300.")])
@@ -463,7 +468,7 @@ def test_ruleset_reference(self):
         )
 
     def test_ruleset_examples(self):
-        cr = ComprehensiveRules.from_text(self.buffer)
+        cr = ComprehensiveRules.parse(self.buffer)
 
         self.assertEqual(len(cr.ruleset["100.2a"].examples), 1)
         self.assertEqual(
@@ -474,26 +479,26 @@ def test_ruleset_examples(self):
         self.assertEqual(len(cr.ruleset["201"].examples), 2)
 
     def test_glossary_access(self):
-        cr = ComprehensiveRules.from_text(self.buffer)
+        cr = ComprehensiveRules.parse(self.buffer)
 
         self.assertEqual(cr.glossary["ipsum"].description, "Integer id ultrices augue.")
         self.assertEqual(cr.glossary["IPSUM"].description, "Integer id ultrices augue.")
         self.assertEqual(cr.glossary["Ipsum"].description, "Integer id ultrices augue.")
 
     def test_glossary_key_error(self):
-        cr = ComprehensiveRules.from_text(self.buffer)
+        cr = ComprehensiveRules.parse(self.buffer)
 
         with self.assertRaises(KeyError):
             print(cr.glossary["not_found"])
 
     def test_search_dont_search_example(self):
-        cr = ComprehensiveRules.from_text(self.buffer)
+        cr = ComprehensiveRules.parse(self.buffer)
 
         found = cr.search("tempor")
         self.assertEqual(len(found), 0)
 
     def test_search(self):
-        cr = ComprehensiveRules.from_text(self.buffer)
+        cr = ComprehensiveRules.parse(self.buffer)
 
         found = cr.search("ipsum")
         self.assertEqual(len(found), 3)
@@ -502,7 +507,7 @@ def test_search(self):
         self.assertIn(cr.glossary["Ipsum"], found)
 
     def test_range_no_up(self):
-        cr = ComprehensiveRules.from_text(self.buffer)
+        cr = ComprehensiveRules.parse(self.buffer)
 
         found = cr.ruleset.range("100")
         self.assertEqual(len(found), 4)
@@ -512,15 +517,15 @@ def test_range_no_up(self):
         self.assertIn(cr.ruleset["100.2a"], found)
 
     def test_range_sub_rule_no_up(self):
-        cr = ComprehensiveRules.from_text(self.buffer)
+        cr = ComprehensiveRules.parse(self.buffer)
 
         found = cr.ruleset.range("100.2")
         self.assertEqual(len(found), 2)
         self.assertIn(cr.ruleset["100.2"], found)
         self.assertIn(cr.ruleset["100.2a"], found)
 
     def test_range_sub_rule(self):
-        cr = ComprehensiveRules.from_text(self.buffer)
+        cr = ComprehensiveRules.parse(self.buffer)
 
         found = cr.ruleset.range("100", "300")
         self.assertEqual(len(found), 7)
@@ -534,20 +539,29 @@ def test_range_sub_rule(self):
         self.assertNotIn(cr.ruleset["300"], found)
 
 
-class TestComprehensiveRuleReal(unittest.TestCase):
-    def test_resolve_latest(self):
-        """
-        This test is some sort of integration test, we need to ensure that
-        this feature is not broken
-        """
-        url = ComprehensiveRules.latest()
-        self.assertRegex(url, r"https://media.wizards.com/.*/MagicComp.+\.txt")
+@pytest.mark.asyncio
+class TestRuleExplorer(unittest.IsolatedAsyncioTestCase):
+    @pytest.mark.skip_remote_api_2
+    async def test_resolve_latest(self):
+        explorer = RuleExplorer()
+        url = await explorer.latest()
+        self.assertRegex(str(url), r"https://media.wizards.com/.*/MagicComp.+\.txt")
 
-    def test_real_rules_have_a_bunch_of_data(self):
+    async def test_real_rules_have_a_bunch_of_data(self):
+        explorer = RuleExplorer()
         path = os.path.join(os.path.dirname(__file__), "rule.20230203.txt")
         self.assertTrue(os.path.exists(path))
 
-        rule = ComprehensiveRules.from_file(path)
+        rule = await explorer.open(path)
         self.assertEqual(rule.effective.date, datetime.date(2023, 2, 3))
         self.assertGreater(len(rule.ruleset), 2800)
         self.assertGreater(len(rule.glossary), 200)
+
+    @pytest.mark.skip_remote_api_2
+    async def test_open_latest_remote(self):
+        explorer = RuleExplorer()
+
+        rule = await explorer.open()
+        self.assertGreaterEqual(rule.effective.date, datetime.date(2023, 2, 3))
+        self.assertGreater(len(rule.ruleset), 2800)
+        self.assertGreater(len(rule.glossary), 200)