Spaces:
Sleeping
Sleeping
Compare MAX channel handles without case changes affecting post identity
Browse files- max_api.py +17 -2
- tests/test_max_api.py +19 -0
max_api.py
CHANGED
|
@@ -91,6 +91,22 @@ def message_id_from_link(channel_id: int, post_id: str) -> str:
|
|
| 91 |
return "mid." + chat_bytes.hex() + message_bytes.hex()
|
| 92 |
|
| 93 |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 94 |
def get_public_post(session, slug, post_id, canonical_link, timeout):
|
| 95 |
title = slug
|
| 96 |
try:
|
|
@@ -129,8 +145,7 @@ def get_public_post(session, slug, post_id, canonical_link, timeout):
|
|
| 129 |
LOGGER.warning("MAX API returned a different message for %s", canonical_link)
|
| 130 |
return title, None
|
| 131 |
if message.get("url"):
|
| 132 |
-
|
| 133 |
-
if returned_path != urlsplit(canonical_link).path.rstrip("/"):
|
| 134 |
LOGGER.warning("MAX API returned a different post URL for %s", canonical_link)
|
| 135 |
return title, None
|
| 136 |
views = stat.get("views")
|
|
|
|
| 91 |
return "mid." + chat_bytes.hex() + message_bytes.hex()
|
| 92 |
|
| 93 |
|
| 94 |
+
def same_public_post_url(returned_url, canonical_link):
|
| 95 |
+
"""Channel handles are case-insensitive; short message IDs are not."""
|
| 96 |
+
try:
|
| 97 |
+
actual, expected = urlsplit(returned_url), urlsplit(canonical_link)
|
| 98 |
+
if (actual.scheme not in ("http", "https") or actual.hostname != "max.ru"
|
| 99 |
+
or actual.username or actual.password or actual.port):
|
| 100 |
+
return False
|
| 101 |
+
actual_parts = actual.path.rstrip("/").split("/")
|
| 102 |
+
expected_parts = expected.path.rstrip("/").split("/")
|
| 103 |
+
return (len(actual_parts) == len(expected_parts) == 3
|
| 104 |
+
and actual_parts[1].casefold() == expected_parts[1].casefold()
|
| 105 |
+
and actual_parts[2] == expected_parts[2])
|
| 106 |
+
except (ValueError, TypeError):
|
| 107 |
+
return False
|
| 108 |
+
|
| 109 |
+
|
| 110 |
def get_public_post(session, slug, post_id, canonical_link, timeout):
|
| 111 |
title = slug
|
| 112 |
try:
|
|
|
|
| 145 |
LOGGER.warning("MAX API returned a different message for %s", canonical_link)
|
| 146 |
return title, None
|
| 147 |
if message.get("url"):
|
| 148 |
+
if not same_public_post_url(message["url"], canonical_link):
|
|
|
|
| 149 |
LOGGER.warning("MAX API returned a different post URL for %s", canonical_link)
|
| 150 |
return title, None
|
| 151 |
views = stat.get("views")
|
tests/test_max_api.py
CHANGED
|
@@ -79,6 +79,25 @@ class MaxPublicPostTests(unittest.TestCase):
|
|
| 79 |
session, "vkontakte", "AaE-viFyB3E", URL, (5, 8),
|
| 80 |
)[1], 0)
|
| 81 |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 82 |
def test_output_keeps_max_counters_and_partial_sum(self):
|
| 83 |
lines = [("max", URL, "vkontakte", "AaE-viFyB3E"),
|
| 84 |
("max", "https://max.ru/other/AaE-viFyB3E", "other", "AaE-viFyB3E")]
|
|
|
|
| 79 |
session, "vkontakte", "AaE-viFyB3E", URL, (5, 8),
|
| 80 |
)[1], 0)
|
| 81 |
|
| 82 |
+
@patch.dict(os.environ, {"MAX_ACCESS_TOKEN": "test-token"})
|
| 83 |
+
def test_case_only_channel_difference_keeps_exact_message_check(self):
|
| 84 |
+
session = Mock()
|
| 85 |
+
session.get.side_effect = [page_response(), Mock(status_code=200, json=lambda: {
|
| 86 |
+
"body": {"mid": MID}, "stat": {"views": 1197},
|
| 87 |
+
"url": "https://max.ru/VKONTAKTE/AaE-viFyB3E/",
|
| 88 |
+
})]
|
| 89 |
+
self.assertEqual(max_api.get_public_post(session, "vkontakte", "AaE-viFyB3E", URL, (5, 8))[1], 1197)
|
| 90 |
+
|
| 91 |
+
def test_url_comparison_rejects_other_posts_and_hosts(self):
|
| 92 |
+
for url in ["https://max.ru/vkontakte/aaE-viFyB3E",
|
| 93 |
+
"https://max.ru/other/AaE-viFyB3E",
|
| 94 |
+
"https://evil.test/vkontakte/AaE-viFyB3E",
|
| 95 |
+
"https://user:pass@max.ru/vkontakte/AaE-viFyB3E",
|
| 96 |
+
"https://max.ru:8443/vkontakte/AaE-viFyB3E"]:
|
| 97 |
+
with self.subTest(url=url):
|
| 98 |
+
self.assertFalse(max_api.same_public_post_url(url, URL))
|
| 99 |
+
self.assertTrue(max_api.same_public_post_url("https://max.ru/Vkontakte/AaE-viFyB3E?from=share", URL))
|
| 100 |
+
|
| 101 |
def test_output_keeps_max_counters_and_partial_sum(self):
|
| 102 |
lines = [("max", URL, "vkontakte", "AaE-viFyB3E"),
|
| 103 |
("max", "https://max.ru/other/AaE-viFyB3E", "other", "AaE-viFyB3E")]
|