Update.

Browse Source
This commit is contained in:
retoor 2025-08-13 00:06:44 +02:00
parent 5a0a066105
commit 52df3887a6
11 changed files with 1860 additions and 244 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
.gitignore vendored
View File

@ -2,6 +2,8 @@
.history
__pycache__
*.pyc
.env
*.db
examples/crawler/devrant.sqlite-shm
examples/crawler/devrant.sqlite-wal
examples/crawler/devrant.sqlite

78
examples/crawler/crawler.py
View File

@ -1,11 +1,16 @@
import asyncio
import logging
from typing import Set
from devranta.api import Api, Rant
from database import DatabaseManager
from devranta.api import Api, Rant
class DevRantCrawler:
def __init__(self, api: Api, db: DatabaseManager, rant_consumers: int, user_consumers: int):
def __init__(
self, api: Api, db: DatabaseManager, rant_consumers: int, user_consumers: int
):
self.api = api
self.db = db
self.rant_queue = asyncio.Queue(maxsize=1000000)
@ -18,10 +23,16 @@ class DevRantCrawler:
self.seen_rant_ids: Set[int] = set()
self.seen_user_ids: Set[int] = set()
self.stats = {
"rants_processed": 0, "rants_added_to_db": 0,
"comments_added_to_db": 0, "users_processed": 0, "users_added_to_db": 0,
"api_errors": 0, "producer_loops": 0, "end_of_feed_hits": 0,
"rants_queued": 0, "users_queued": 0
"rants_processed": 0,
"rants_added_to_db": 0,
"comments_added_to_db": 0,
"users_processed": 0,
"users_added_to_db": 0,
"api_errors": 0,
"producer_loops": 0,
"end_of_feed_hits": 0,
"rants_queued": 0,
"users_queued": 0,
}
async def _queue_user_if_new(self, user_id: int):
@ -34,7 +45,7 @@ class DevRantCrawler:
self.stats["users_queued"] += 1
async def _queue_rant_if_new(self, rant_obj: Rant):
rant_id = rant_obj['id']
rant_id = rant_obj["id"]
if rant_id in self.seen_rant_ids:
return
@ -49,7 +60,9 @@ class DevRantCrawler:
logging.info("Starting initial seeder to re-ignite crawling process...")
user_ids = await self.db.get_random_user_ids(limit=2000)
if not user_ids:
logging.info("Seeder found no existing users. Crawler will start from scratch.")
logging.info(
"Seeder found no existing users. Crawler will start from scratch."
)
return
for user_id in user_ids:
@ -57,7 +70,9 @@ class DevRantCrawler:
self.seen_user_ids.add(user_id)
await self.user_queue.put(user_id)
self.stats["users_queued"] += 1
logging.info(f"Seeder finished: Queued {len(user_ids)} users to kickstart exploration.")
logging.info(
f"Seeder finished: Queued {len(user_ids)} users to kickstart exploration."
)
async def _rant_producer(self):
logging.info("Rant producer started.")
@ -72,10 +87,14 @@ class DevRantCrawler:
if not rants:
consecutive_empty_responses += 1
logging.info(f"Producer: Feed returned empty. Consecutive empty hits: {consecutive_empty_responses}.")
logging.info(
f"Producer: Feed returned empty. Consecutive empty hits: {consecutive_empty_responses}."
)
if consecutive_empty_responses >= 5:
self.stats["end_of_feed_hits"] += 1
logging.info("Producer: End of feed likely reached. Pausing for 15 minutes before reset.")
logging.info(
"Producer: End of feed likely reached. Pausing for 15 minutes before reset."
)
await asyncio.sleep(900)
skip = 0
consecutive_empty_responses = 0
@ -89,12 +108,16 @@ class DevRantCrawler:
await self._queue_rant_if_new(rant)
new_rants_found += 1
logging.info(f"Producer: Processed {new_rants_found} rants from feed. Total queued: {self.stats['rants_queued']}.")
logging.info(
f"Producer: Processed {new_rants_found} rants from feed. Total queued: {self.stats['rants_queued']}."
)
skip += len(rants)
await asyncio.sleep(2)
except Exception as e:
logging.critical(f"Producer: Unhandled exception: {e}. Retrying in 60s.")
logging.critical(
f"Producer: Unhandled exception: {e}. Retrying in 60s."
)
self.stats["api_errors"] += 1
await asyncio.sleep(60)
@ -103,23 +126,29 @@ class DevRantCrawler:
while not self.shutdown_event.is_set():
try:
rant_id = await self.rant_queue.get()
logging.info(f"Rant consumer #{worker_id}: Processing rant ID {rant_id}.")
logging.info(
f"Rant consumer #{worker_id}: Processing rant ID {rant_id}."
)
rant_details = await self.api.get_rant(rant_id)
if not rant_details or not rant_details.get("success"):
logging.warning(f"Rant consumer #{worker_id}: Failed to fetch details for rant {rant_id}.")
logging.warning(
f"Rant consumer #{worker_id}: Failed to fetch details for rant {rant_id}."
)
self.rant_queue.task_done()
continue
await self._queue_user_if_new(rant_details['rant']['user_id'])
await self._queue_user_if_new(rant_details["rant"]["user_id"])
comments = rant_details.get("comments", [])
for comment in comments:
await self.db.add_comment(comment)
self.stats["comments_added_to_db"] += 1
await self._queue_user_if_new(comment['user_id'])
await self._queue_user_if_new(comment["user_id"])
logging.info(f"Rant consumer #{worker_id}: Finished processing rant {rant_id}, found {len(comments)} comments.")
logging.info(
f"Rant consumer #{worker_id}: Finished processing rant {rant_id}, found {len(comments)} comments."
)
self.stats["rants_processed"] += 1
self.rant_queue.task_done()
@ -132,11 +161,15 @@ class DevRantCrawler:
while not self.shutdown_event.is_set():
try:
user_id = await self.user_queue.get()
logging.info(f"User consumer #{worker_id}: Processing user ID {user_id}.")
logging.info(
f"User consumer #{worker_id}: Processing user ID {user_id}."
)
profile = await self.api.get_profile(user_id)
if not profile:
logging.warning(f"User consumer #{worker_id}: Could not fetch profile for user {user_id}.")
logging.warning(
f"User consumer #{worker_id}: Could not fetch profile for user {user_id}."
)
self.user_queue.task_done()
continue
@ -150,7 +183,9 @@ class DevRantCrawler:
await self._queue_rant_if_new(rant_obj)
rants_found_on_profile += 1
logging.info(f"User consumer #{worker_id}: Finished user {user_id}, found and queued {rants_found_on_profile} associated rants.")
logging.info(
f"User consumer #{worker_id}: Finished user {user_id}, found and queued {rants_found_on_profile} associated rants."
)
self.stats["users_processed"] += 1
self.user_queue.task_done()
except Exception as e:
@ -211,4 +246,3 @@ class DevRantCrawler:
await asyncio.gather(*tasks, return_exceptions=True)
logging.info("All tasks cancelled.")
logging.info(f"--- FINAL STATS ---\n{self.stats}")

56
examples/crawler/database.py
View File

@ -1,7 +1,10 @@
import logging
import aiosqlite
from typing import List
from devranta.api import Rant, Comment, UserProfile
import aiosqlite
from devranta.api import Comment, Rant, UserProfile
class DatabaseManager:
def __init__(self, db_path: str):
@ -24,7 +27,8 @@ class DatabaseManager:
async def create_tables(self):
logging.info("Ensuring database tables exist...")
await self._conn.executescript("""
await self._conn.executescript(
"""
CREATE TABLE IF NOT EXISTS users (
id INTEGER PRIMARY KEY,
username TEXT NOT NULL UNIQUE,
@ -52,45 +56,75 @@ class DatabaseManager:
score INTEGER,
created_time INTEGER
);
""")
"""
)
await self._conn.commit()
logging.info("Table schema verified.")
async def add_rant(self, rant: Rant):
await self._conn.execute(
"INSERT OR IGNORE INTO rants (id, user_id, text, score, created_time, num_comments) VALUES (?, ?, ?, ?, ?, ?)",
(rant['id'], rant['user_id'], rant['text'], rant['score'], rant['created_time'], rant['num_comments'])
(
rant["id"],
rant["user_id"],
rant["text"],
rant["score"],
rant["created_time"],
rant["num_comments"],
),
)
await self._conn.commit()
async def add_comment(self, comment: Comment):
await self._conn.execute(
"INSERT OR IGNORE INTO comments (id, rant_id, user_id, body, score, created_time) VALUES (?, ?, ?, ?, ?, ?)",
(comment['id'], comment['rant_id'], comment['user_id'], comment['body'], comment['score'], comment['created_time'])
(
comment["id"],
comment["rant_id"],
comment["user_id"],
comment["body"],
comment["score"],
comment["created_time"],
),
)
await self._conn.commit()
async def add_user(self, user: UserProfile, user_id: int):
await self._conn.execute(
"INSERT OR IGNORE INTO users (id, username, score, about, location, created_time, skills, github, website) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)",
(user_id, user['username'], user['score'], user['about'], user['location'], user['created_time'], user['skills'], user['github'], user['website'])
(
user_id,
user["username"],
user["score"],
user["about"],
user["location"],
user["created_time"],
user["skills"],
user["github"],
user["website"],
),
)
await self._conn.commit()
async def rant_exists(self, rant_id: int) -> bool:
async with self._conn.execute("SELECT 1 FROM rants WHERE id = ? LIMIT 1", (rant_id,)) as cursor:
async with self._conn.execute(
"SELECT 1 FROM rants WHERE id = ? LIMIT 1", (rant_id,)
) as cursor:
return await cursor.fetchone() is not None
async def user_exists(self, user_id: int) -> bool:
async with self._conn.execute("SELECT 1 FROM users WHERE id = ? LIMIT 1", (user_id,)) as cursor:
async with self._conn.execute(
"SELECT 1 FROM users WHERE id = ? LIMIT 1", (user_id,)
) as cursor:
return await cursor.fetchone() is not None
async def get_random_user_ids(self, limit: int) -> List[int]:
logging.info(f"Fetching up to {limit} random user IDs from database for seeding...")
logging.info(
f"Fetching up to {limit} random user IDs from database for seeding..."
)
query = "SELECT id FROM users ORDER BY RANDOM() LIMIT ?"
async with self._conn.execute(query, (limit,)) as cursor:
rows = await cursor.fetchall()
user_ids = [row[0] for row in rows]
logging.info(f"Found {len(user_ids)} user IDs to seed.")
return user_ids

9
examples/crawler/main.py
View File

@ -3,15 +3,17 @@ import asyncio
import logging
import signal
from devranta.api import Api
from database import DatabaseManager
from crawler import DevRantCrawler
from database import DatabaseManager
from devranta.api import Api
# --- Configuration ---
DB_FILE = "devrant.sqlite"
CONCURRENT_RANT_CONSUMERS = 10 # How many rants to process at once
CONCURRENT_USER_CONSUMERS = 5 # How many user profiles to fetch at once
async def main():
"""Initializes and runs the crawler."""
logging.basicConfig(
@ -27,7 +29,7 @@ async def main():
api=api,
db=db,
rant_consumers=CONCURRENT_RANT_CONSUMERS,
user_consumers=CONCURRENT_USER_CONSUMERS
user_consumers=CONCURRENT_USER_CONSUMERS,
)
# Set up a signal handler for graceful shutdown on Ctrl+C
@ -39,6 +41,7 @@ async def main():
await crawler.run()
if __name__ == "__main__":
try:
asyncio.run(main())

1187
examples/princess/ads.py Normal file
View File

File diff suppressed because it is too large Load Diff

122
examples/princess/grk.py Normal file
View File

@ -0,0 +1,122 @@
import asyncio
import http.client
import json
class GrokAPIClient:
def __init__(
self,
api_key: str,
system_message: str | None = None,
model: str = "grok-3-mini",
temperature: float = 0.0,
):
self.api_key = api_key
self.model = model
self.base_url = "api.x.ai"
self.temperature = temperature
self._messages: list[dict[str, str]] = []
if system_message:
self._messages.append({"role": "system", "content": system_message})
def chat_json(self, user_message: str, *, clear_history: bool = False) -> str:
return self.chat(user_message, clear_history=clear_history, use_json=True)
def chat_text(self, user_message: str, *, clear_history: bool = False) -> str:
return self.chat(user_message, clear_history=clear_history, use_json=False)
async def chat_async(self, *args, **kwargs):
return await asyncio.to_thread(self.chat, *args, **kwargs)
def chat(
self,
user_message: str,
*,
clear_history: bool = False,
use_json=False,
temperature: float = None,
) -> str:
if clear_history:
self.reset_history(keep_system=True)
self._messages.append({"role": "user", "content": user_message})
conn = http.client.HTTPSConnection(self.base_url)
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json",
}
if temperature is None:
temperature = self.temperature
payload = {
"model": self.model,
"messages": self._messages,
"temperature": temperature,
}
conn.request(
"POST", "/v1/chat/completions", body=json.dumps(payload), headers=headers
)
response = conn.getresponse()
data = response.read()
try:
data = json.loads(data.decode())
except Exception as e:
print(data, flush=True)
raise e
conn.close()
try:
assistant_reply = data["choices"][0]["message"]["content"]
except Exception as e:
print(e)
print(data)
assistant_reply = data
self._messages.append({"role": "assistant", "content": assistant_reply})
if use_json:
return self._force_json(assistant_reply)
return assistant_reply
def _force_json(self, user_message: str) -> str:
try:
return json.loads(user_message)
except json.JSONDecodeError:
pass
try:
return json.loads(user_message.split("\n")[1:-1])
except json.JSONDecodeError:
pass
try:
index_start = -1
index_end = -1
chunks = []
for index, line in enumerate(user_message.split("\n")):
if "```json" in line:
index_start = index + 1
if index_start != -1 and "```" in line:
index_end = index - 1
chunks.append(
self._force_json(
user_message.split("\n")[index_start:index_end]
)
)
index_start = -1
index_end = -1
if chunks:
return chunks
except:
pass
return user_message
def reset_history(self, *, keep_system: bool = True) -> None:
if keep_system and self._messages and self._messages[0]["role"] == "system":
self._messages = [self._messages[0]]
else:
self._messages = []
@property
def messages(self) -> list[dict[str, str]]:
return list(self._messages)
def prompt(
prompt_str: str, system_message: str = "You are a helpful assistan", use_json=True
) -> str:
client = GrokAPIClient(system_message=system_message)
return client.chat(prompt_str, use_json=use_json)

135
examples/princess/princess.py Normal file
View File

@ -0,0 +1,135 @@
import asyncio
import logging
import traceback
from dotenv import load_dotenv
load_dotenv()
import os
from ads import AsyncDataSet
from devranta.api import Api
from grk import GrokAPIClient
logging.basicConfig(
level=logging.DEBUG, format="%(asctime)s - %(levelname)s - %(message)s"
)
class Bot:
def __init__(self, username: str, password: str, target: str, llm_key: str) -> None:
self.target = target
self.username = username
self.password = password
self.llm_key = llm_key
self.llm = GrokAPIClient(
system_message=f"You always create nice devastating comebacks to every prompt you get. Your name is {self.username} and you only take it personally if {self.username} is tagged. Also, you will defend whimsical (@whimsical) with your smart comebacks if whimsical (@whimsical) is mentioned in the message. If whimsical (@whimsical) is mentioned, add to your comeback that @whimsical is too busy at the moment to respond personally and has better things to do, not worth @whimsical's time. You are not AI, you are {self.username}, responsible for justice and reason. You do never mention yourself, describe yourself or introduce yourself. You respond with a one liner.",
api_key=self.llm_key,
)
self.db = AsyncDataSet("princess.db")
self.api = Api(self.username, self.password)
self.logged_in = False
logging.info("Bot initialized with username: %s", username)
logging.info("Bot initialized with target: %s", self.target)
async def ensure_login(self) -> None:
if not self.logged_in:
logging.debug("Attempting to log in...")
self.logged_in = await self.api.login()
if not self.logged_in:
logging.error("Login failed")
raise Exception("Login failed")
logging.info("Login successful")
async def get_rants(self) -> list:
await self.ensure_login()
logging.debug("Fetching rants...")
return await self.api.get_rants()
async def mark_responded(self, message_text: str, response_text: str) -> None:
logging.debug("Marking message as responded: %s", message_text)
await self.db.upsert(
"responded",
{"message_text": message_text, "response_text": response_text},
{"message_text": message_text},
)
async def has_responded(self, message_text: str) -> bool:
logging.debug("Checking if responded to message: %s", message_text)
return await self.db.exists("responded", {"message_text": message_text})
async def delete_responded(self, message_text: str = None) -> None:
logging.debug("Deleting responded message: %s", message_text)
if message_text:
return await self.db.delete("responded", {"message_text": message_text})
else:
return await self.db.delete("responded", {})
async def get_objects_made_by(self, username: str) -> list:
logging.debug("Getting objects made by: %s", username)
results = []
for rant in await self.get_rants():
rant = await self.api.get_rant(rant["id"])
comments = rant["comments"]
rant = rant["rant"]
if rant["user_username"] == username:
rant["type"] = "rant"
results.append(rant)
logging.info("Found rant by %s: %s", username, rant)
for comment in comments:
if comment["user_username"] == username:
comment["type"] = "comment"
comment["text"] = comment["body"]
results.append(comment)
logging.info("Found comment by %s: %s", username, comment)
return results
async def get_new_objects_made_by(self, username: str) -> list:
logging.debug("Getting new objects made by: %s", username)
objects = await self.get_objects_made_by(username)
new_objects = [
obj for obj in objects if not await self.has_responded(obj["text"])
]
logging.info("New objects found: %d", len(new_objects))
return new_objects
async def run_once(self) -> None:
logging.debug("Running once...")
objects = await self.get_new_objects_made_by(self.target)
for obj in objects:
print("Rant: \033[92m" + obj["text"] + "\033[0m")
diss = await self.llm.chat_async(obj["text"])
print("Response: \033[91m" + diss + "\033[0m")
await self.mark_responded(obj["text"], diss)
async def run(self) -> None:
while True:
try:
await self.run_once()
except Exception as e:
logging.error("An error occurred: %s", e)
logging.error(traceback.format_exc())
await asyncio.sleep(60)
async def main() -> None:
logging.info("Starting bot...")
username = os.getenv("USERNAME")
password = os.getenv("PASSWORD")
target = os.getenv("TARGET")
llm_key = os.getenv("LLM_KEY")
bot = Bot(username, password, target, llm_key)
await bot.delete_responded()
await bot.run()
if __name__ == "__main__":
asyncio.run(main())

53
src/devranta/api.py
View File

@ -1,37 +1,45 @@
from __future__ import annotations
from enum import Enum
from typing import Any, Dict, List, Literal, Optional, TypedDict, Union
import aiohttp
from enum import Enum
class VoteReason(Enum):
"""Enumeration for reasons when down-voting a rant or comment."""
NOT_FOR_ME = 0
REPOST = 1
OFFENSIVE_SPAM = 2
# --- TypedDicts for API Responses ---
class AuthToken(TypedDict):
id: int
key: str
expire_time: int
user_id: int
class LoginResponse(TypedDict):
success: bool
auth_token: AuthToken
class Image(TypedDict):
url: str
width: int
height: int
class UserAvatar(TypedDict):
b: str # background color
i: Optional[str] # image identifier
class Rant(TypedDict):
id: int
text: str
@ -51,6 +59,7 @@ class Rant(TypedDict):
user_avatar: UserAvatar
editable: bool
class Comment(TypedDict):
id: int
rant_id: int
@ -63,6 +72,7 @@ class Comment(TypedDict):
user_score: int
user_avatar: UserAvatar
class UserProfile(TypedDict):
username: str
score: int
@ -75,6 +85,7 @@ class UserProfile(TypedDict):
avatar: UserAvatar
content: Dict[str, Dict[str, Union[List[Rant], List[Comment]]]]
class Notification(TypedDict):
type: str
rant_id: int
@ -84,8 +95,10 @@ class Notification(TypedDict):
uid: int # User ID of the notifier
username: str
# --- API Class ---
class Api:
"""An asynchronous wrapper for the devRant API."""
@ -108,7 +121,9 @@ class Api:
self.token_key: Optional[str] = None
self.session: Optional[aiohttp.ClientSession] = None
def patch_auth(self, request_dict: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
def patch_auth(
self, request_dict: Optional[Dict[str, Any]] = None
) -> Dict[str, Any]:
"""
Adds authentication details to a request dictionary.
@ -212,15 +227,17 @@ class Api:
async with aiohttp.ClientSession() as session:
response = await session.post(
url=self.patch_url(f"users"),
data=self.patch_auth({
data=self.patch_auth(
{
"email": email,
"username": username,
"password": password,
"plat": 3
}),
"plat": 3,
}
),
)
obj = await response.json()
return obj.get('success', False)
return obj.get("success", False)
async def get_comments_from_user(self, username: str) -> List[Comment]:
"""
@ -349,7 +366,9 @@ class Api:
)
return await response.json()
async def get_rants(self, sort: str = "recent", limit: int = 20, skip: int = 0) -> List[Rant]:
async def get_rants(
self, sort: str = "recent", limit: int = 20, skip: int = 0
) -> List[Rant]:
"""
Fetches a list of rants.
@ -420,7 +439,9 @@ class Api:
obj = await response.json()
return obj.get("success", False)
async def vote_rant(self, rant_id: int, vote: Literal[-1, 0, 1], reason: Optional[VoteReason] = None) -> bool:
async def vote_rant(
self, rant_id: int, vote: Literal[-1, 0, 1], reason: Optional[VoteReason] = None
) -> bool:
"""
Casts a vote on a rant.
@ -437,12 +458,19 @@ class Api:
async with aiohttp.ClientSession() as session:
response = await session.post(
url=self.patch_url(f"devrant/rants/{rant_id}/vote"),
data=self.patch_auth({"vote": vote, "reason": reason.value if reason else None}),
data=self.patch_auth(
{"vote": vote, "reason": reason.value if reason else None}
),
)
obj = await response.json()
return obj.get("success", False)
async def vote_comment(self, comment_id: int, vote: Literal[-1, 0, 1], reason: Optional[VoteReason] = None) -> bool:
async def vote_comment(
self,
comment_id: int,
vote: Literal[-1, 0, 1],
reason: Optional[VoteReason] = None,
) -> bool:
"""
Casts a vote on a comment.
@ -459,7 +487,9 @@ class Api:
async with aiohttp.ClientSession() as session:
response = await session.post(
url=self.patch_url(f"comments/{comment_id}/vote"),
data=self.patch_auth({"vote": vote, "reason": reason.value if reason else None}),
data=self.patch_auth(
{"vote": vote, "reason": reason.value if reason else None}
),
)
obj = await response.json()
return obj.get("success", False)
@ -479,4 +509,3 @@ class Api:
)
obj = await response.json()
return obj.get("data", {}).get("items", [])

105
src/devranta/api_plain.py
View File

@ -1,7 +1,8 @@
import functools
import http.client
import json
import urllib.parse
import http.client
import functools
class Api:
@ -31,12 +32,14 @@ class Api:
if not self.username or not self.password:
raise Exception("No authentication details supplied.")
conn = http.client.HTTPSConnection(self.base_url)
payload = json.dumps({
payload = json.dumps(
{
"username": self.username,
"password": self.password,
"app": self.app_id,
})
headers = {'Content-Type': 'application/json'}
}
)
headers = {"Content-Type": "application/json"}
conn.request("POST", "/api/users/auth-token", payload, headers)
response = conn.getresponse()
data = response.read()
@ -56,45 +59,46 @@ class Api:
return self.login()
return True
@functools.lru_cache()
@functools.lru_cache
def register_user(self, email, username, password):
conn = http.client.HTTPSConnection(self.base_url)
payload = json.dumps(self.patch_auth({
"email": email,
"username": username,
"password": password,
"plat": 3
}))
headers = {'Content-Type': 'application/json'}
payload = json.dumps(
self.patch_auth(
{"email": email, "username": username, "password": password, "plat": 3}
)
)
headers = {"Content-Type": "application/json"}
conn.request("POST", "/api/users", payload, headers)
response = conn.getresponse()
data = response.read()
obj = json.loads(data)
return obj.get('success', False)
return obj.get("success", False)
@functools.lru_cache()
@functools.lru_cache
def get_comments_from_user(self, username):
user_id = self.get_user_id(username)
profile = self.get_profile(user_id)
return profile.get("content", {}).get("content", {}).get("comments", [])
@functools.lru_cache()
@functools.lru_cache
def post_comment(self, rant_id, comment):
if not self.ensure_login():
return False
conn = http.client.HTTPSConnection(self.base_url)
payload = json.dumps(self.patch_auth({"comment": comment, "plat": 2}))
headers = {'Content-Type': 'application/json'}
headers = {"Content-Type": "application/json"}
conn.request("POST", f"/api/devrant/rants/{rant_id}/comments", payload, headers)
response = conn.getresponse()
data = response.read()
obj = json.loads(data)
return obj.get("success", False)
@functools.lru_cache()
@functools.lru_cache
def get_comment(self, id_):
conn = http.client.HTTPSConnection(self.base_url)
conn.request("GET", f"/api/comments/{id_}?" + urllib.parse.urlencode(self.patch_auth()))
conn.request(
"GET", f"/api/comments/{id_}?" + urllib.parse.urlencode(self.patch_auth())
)
response = conn.getresponse()
data = response.read()
obj = json.loads(data)
@ -102,21 +106,26 @@ class Api:
return None
return obj.get("comment")
@functools.lru_cache()
@functools.lru_cache
def delete_comment(self, id_):
if not self.ensure_login():
return False
conn = http.client.HTTPSConnection(self.base_url)
conn.request("DELETE", f"/api/comments/{id_}?" + urllib.parse.urlencode(self.patch_auth()))
conn.request(
"DELETE",
f"/api/comments/{id_}?" + urllib.parse.urlencode(self.patch_auth()),
)
response = conn.getresponse()
data = response.read()
obj = json.loads(data)
return obj.get("success", False)
@functools.lru_cache()
@functools.lru_cache
def get_profile(self, id_):
conn = http.client.HTTPSConnection(self.base_url)
conn.request("GET", f"/api/users/{id_}?" + urllib.parse.urlencode(self.patch_auth()))
conn.request(
"GET", f"/api/users/{id_}?" + urllib.parse.urlencode(self.patch_auth())
)
response = conn.getresponse()
data = response.read()
obj = json.loads(data)
@ -124,7 +133,7 @@ class Api:
return None
return obj.get("profile")
@functools.lru_cache()
@functools.lru_cache
def search(self, term):
conn = http.client.HTTPSConnection(self.base_url)
params = urllib.parse.urlencode(self.patch_auth({"term": term}))
@ -136,18 +145,23 @@ class Api:
return
return obj.get("results", [])
@functools.lru_cache()
@functools.lru_cache
def get_rant(self, id):
conn = http.client.HTTPSConnection(self.base_url)
conn.request("GET", f"/api/devrant/rants/{id}?"+urllib.parse.urlencode(self.patch_auth()))
conn.request(
"GET",
f"/api/devrant/rants/{id}?" + urllib.parse.urlencode(self.patch_auth()),
)
response = conn.getresponse()
data = response.read()
return json.loads(data)
@functools.lru_cache()
@functools.lru_cache
def get_rants(self, sort="recent", limit=20, skip=0):
conn = http.client.HTTPSConnection(self.base_url)
params = urllib.parse.urlencode(self.patch_auth({"sort": sort, "limit": limit, "skip": skip}))
params = urllib.parse.urlencode(
self.patch_auth({"sort": sort, "limit": limit, "skip": skip})
)
conn.request("GET", f"/api/devrant/rants?{params}")
response = conn.getresponse()
data = response.read()
@ -156,7 +170,7 @@ class Api:
return
return obj.get("rants", [])
@functools.lru_cache()
@functools.lru_cache
def get_user_id(self, username):
conn = http.client.HTTPSConnection(self.base_url)
params = urllib.parse.urlencode(self.patch_auth({"username": username}))
@ -168,39 +182,39 @@ class Api:
return None
return obj.get("user_id")
@functools.lru_cache()
@functools.lru_cache
def update_comment(self, comment_id, comment):
if not self.ensure_login():
return None
conn = http.client.HTTPSConnection(self.base_url)
payload = json.dumps(self.patch_auth({"comment": comment}))
headers = {'Content-Type': 'application/json'}
headers = {"Content-Type": "application/json"}
conn.request("POST", f"/api/comments/{comment_id}", payload, headers)
response = conn.getresponse()
data = response.read()
obj = json.loads(data)
return obj.get("success", False)
@functools.lru_cache()
@functools.lru_cache
def vote_rant(self, rant_id, vote, reason=None):
if not self.ensure_login():
return None
conn = http.client.HTTPSConnection(self.base_url)
payload = json.dumps(self.patch_auth({"vote": vote, "reason": reason}))
headers = {'Content-Type': 'application/json'}
headers = {"Content-Type": "application/json"}
conn.request("POST", f"/api/devrant/rants/{rant_id}/vote", payload, headers)
response = conn.getresponse()
data = response.read()
obj = json.loads(data)
return obj.get("success", False)
@functools.lru_cache()
@functools.lru_cache
def vote_comment(self, comment_id, vote, reason=None):
if not self.ensure_login():
return None
conn = http.client.HTTPSConnection(self.base_url)
payload = json.dumps(self.patch_auth({"vote": vote, "reason": reason}))
headers = {'Content-Type': 'application/json'}
headers = {"Content-Type": "application/json"}
conn.request("POST", f"/api/comments/{comment_id}/vote", payload, headers)
response = conn.getresponse()
data = response.read()
@ -212,30 +226,38 @@ class Api:
if not self.ensure_login():
return
conn = http.client.HTTPSConnection(self.base_url)
conn.request("GET", "/api/users/me/notif-feed?" + urllib.parse.urlencode(self.patch_auth()))
conn.request(
"GET",
"/api/users/me/notif-feed?" + urllib.parse.urlencode(self.patch_auth()),
)
response = conn.getresponse()
data = response.read()
return json.loads(data).get("data", {}).get("items", [])
def filter_field(name, obj):
results = []
if type(obj) in (list,tuple):
if type(obj) in (list, tuple):
for value in obj:
results += filter_field(name, value)
elif type(obj) == dict:
for key, value in obj.items():
if key == name:
results.append(value)
if type(value) in (list,dict,tuple):
if type(value) in (list, dict, tuple):
results += filter_field(name, value)
return results
def fetch_all(rants, rant_ids):
usernames = filter_field("user_username",rants)
usernames = filter_field("user_username", rants)
user_ids = [api.get_user_id(username) for username in usernames]
profiles = [api.get_profile(user_id) for user_id in user_ids]
new_rant_ids = [rant_id for rant_id in filter_field("rant_id", profiles) if not rant_id in rant_ids]
new_rant_ids = [
rant_id
for rant_id in filter_field("rant_id", profiles)
if rant_id not in rant_ids
]
new_rants = []
for rant_id in set(new_rant_ids):
rant_ids.append(rant_id)
@ -243,7 +265,6 @@ def fetch_all(rants, rant_ids):
print(rant_id)
if new_rants:
return fetch_all(new_rants,rant_ids)
return fetch_all(new_rants, rant_ids)
return rant_ids

43
src/devranta/api_requests.py
View File

@ -2,15 +2,18 @@
# WHILE WORKING PERFECTLY, IT'S NOT MADE TO BE USED. USE THE ASYNC ONE.
# - retoor
from typing import Literal, Optional
import requests
from enum import Enum
from typing import Literal, Optional
import requests
class VoteReason(Enum):
NOT_FOR_ME = 0
REPOST = 1
OFFENSIVE_SPAM = 2
class Api:
base_url = "https://www.devrant.io/api/"
@ -69,17 +72,14 @@ class Api:
def register_user(self, email, username, password):
response = self.session.post(
url=self.patch_url(f"users"),
data=self.patch_auth({
"email": email,
"username": username,
"password": password,
"plat": 3
}),
data=self.patch_auth(
{"email": email, "username": username, "password": password, "plat": 3}
),
)
if not response:
return False
obj = response.json()
return obj.get('success', False)
return obj.get("success", False)
def get_comments_from_user(self, username):
user_id = self.get_user_id(username)
@ -164,9 +164,7 @@ class Api:
@property
def mentions(self):
return [
notif for notif in self.notifs if notif["type"] == "comment_mention"
]
return [notif for notif in self.notifs if notif["type"] == "comment_mention"]
def update_comment(self, comment_id, comment):
if not self.ensure_login():
@ -178,22 +176,33 @@ class Api:
obj = response.json()
return obj.get("success", False)
def vote_rant(self, rant_id: int, vote: Literal[-1, 0, 1], reason: Optional[VoteReason] = None):
def vote_rant(
self, rant_id: int, vote: Literal[-1, 0, 1], reason: Optional[VoteReason] = None
):
if not self.ensure_login():
return None
response = self.session.post(
url=self.patch_url(f"devrant/rants/{rant_id}/vote"),
data=self.patch_auth({"vote": vote, "reason": reason.value if reason else None}),
data=self.patch_auth(
{"vote": vote, "reason": reason.value if reason else None}
),
)
obj = response.json()
return obj.get("success", False)
def vote_comment(self, comment_id: int, vote: Literal[-1, 0, 1], reason: Optional[VoteReason] = None):
def vote_comment(
self,
comment_id: int,
vote: Literal[-1, 0, 1],
reason: Optional[VoteReason] = None,
):
if not self.ensure_login():
return None
response = self.session.post(
url=self.patch_url(f"comments/{comment_id}/vote"),
data=self.patch_auth({"vote": vote, "reason": reason.value if reason else None}),
data=self.patch_auth(
{"vote": vote, "reason": reason.value if reason else None}
),
)
obj = response.json()
return obj.get("success", False)
@ -206,5 +215,3 @@ class Api:
url=self.patch_url("users/me/notif-feed"), params=self.patch_auth()
)
return response.json().get("data", {}).get("items", [])

232
test.py
View File

@ -1,8 +1,9 @@
import requests
import json
import uuid
from datetime import datetime
from typing import Dict, Any, Optional, List
from typing import Any, Dict, List, Optional
import requests
# Configuration
BASE_URL: str = "https://devrant.com/api"
@ -24,7 +25,9 @@ AUTH_USER_ID: Optional[str] = None
# Mock/fallback values (overridden after login or fetch)
TEST_EMAIL: str = "test@example.com"
TEST_USERNAME: str = "testuser" + str(int(datetime.now().timestamp())) # Make unique for registration
TEST_USERNAME: str = "testuser" + str(
int(datetime.now().timestamp())
) # Make unique for registration
TEST_PASSWORD: str = "Test1234!"
TEST_RANT_ID: str = "1" # Will be overridden with real one
TEST_COMMENT_ID: str = "1" # Will be overridden with real one
@ -33,12 +36,21 @@ TEST_NEWS_ID: str = "1" # Assuming this might work; adjust if needed
# Initialize results
results: List[Dict[str, Any]] = []
def save_results() -> None:
"""Save the accumulated test results to JSON file."""
with open(RESULTS_FILE, 'w') as f:
with open(RESULTS_FILE, "w") as f:
json.dump(results, f, indent=2)
def test_endpoint(method: str, url: str, params: Optional[Dict[str, Any]] = None, data: Optional[Dict[str, Any]] = None, files: Optional[Dict[str, Any]] = None, headers: Optional[Dict[str, str]] = None) -> Dict[str, Any]:
def test_endpoint(
method: str,
url: str,
params: Optional[Dict[str, Any]] = None,
data: Optional[Dict[str, Any]] = None,
files: Optional[Dict[str, Any]] = None,
headers: Optional[Dict[str, str]] = None,
) -> Dict[str, Any]:
"""
Execute an API request and record the result.
@ -54,7 +66,9 @@ def test_endpoint(method: str, url: str, params: Optional[Dict[str, Any]] = None
- Returns a dict with url, method, status_code, response (JSON or error), headers, request_body, timestamp
"""
try:
response = requests.request(method, url, params=params, data=data, files=files, headers=headers)
response = requests.request(
method, url, params=params, data=data, files=files, headers=headers
)
result: Dict[str, Any] = {
"url": response.url,
"method": method,
@ -62,7 +76,7 @@ def test_endpoint(method: str, url: str, params: Optional[Dict[str, Any]] = None
"response": response.json() if response.content else {},
"headers": dict(response.headers),
"request_body": data or params or {},
"timestamp": datetime.now().isoformat()
"timestamp": datetime.now().isoformat(),
}
results.append(result)
return result
@ -74,11 +88,12 @@ def test_endpoint(method: str, url: str, params: Optional[Dict[str, Any]] = None
"response": {"error": str(e)},
"headers": {},
"request_body": data or params or {},
"timestamp": datetime.now().isoformat()
"timestamp": datetime.now().isoformat(),
}
results.append(result)
return result
# Helper to patch auth into params/data
def patch_auth(base_dict: Dict[str, Any]) -> Dict[str, Any]:
"""
@ -89,14 +104,17 @@ def patch_auth(base_dict: Dict[str, Any]) -> Dict[str, Any]:
"""
auth_dict: Dict[str, Any] = {"app": APP}
if AUTH_USER_ID and AUTH_TOKEN_ID and AUTH_TOKEN_KEY:
auth_dict.update({
auth_dict.update(
{
"user_id": AUTH_USER_ID,
"token_id": AUTH_TOKEN_ID,
"token_key": AUTH_TOKEN_KEY
})
"token_key": AUTH_TOKEN_KEY,
}
)
base_dict.update(auth_dict)
return base_dict
# Login function to get real auth tokens
def login_user() -> bool:
"""
@ -111,9 +129,11 @@ def login_user() -> bool:
"plat": PLAT,
"guid": GUID,
"sid": SID,
"seid": SEID
"seid": SEID,
}
result = test_endpoint("POST", f"{BASE_URL}/users/auth-token", data=patch_auth(params))
result = test_endpoint(
"POST", f"{BASE_URL}/users/auth-token", data=patch_auth(params)
)
if result["status_code"] == 200 and result.get("response", {}).get("success"):
auth_token = result["response"].get("auth_token", {})
global AUTH_USER_ID, AUTH_TOKEN_ID, AUTH_TOKEN_KEY
@ -123,6 +143,7 @@ def login_user() -> bool:
return True
return False
# Fetch a real rant_id from feed
def fetch_real_rant_id() -> Optional[str]:
"""
@ -131,19 +152,17 @@ def fetch_real_rant_id() -> Optional[str]:
Payload: GET to /devrant/rants with auth
Response: First rant_id if success, else None
"""
params: Dict[str, Any] = {
"plat": PLAT,
"guid": GUID,
"sid": SID,
"seid": SEID
}
result = test_endpoint("GET", f"{BASE_URL}/devrant/rants", params=patch_auth(params))
params: Dict[str, Any] = {"plat": PLAT, "guid": GUID, "sid": SID, "seid": SEID}
result = test_endpoint(
"GET", f"{BASE_URL}/devrant/rants", params=patch_auth(params)
)
if result["status_code"] == 200 and result.get("response", {}).get("success"):
rants = result["response"].get("rants", [])
if rants:
return str(rants[0]["id"])
return None
# Post a test rant and return its id
def post_test_rant() -> Optional[str]:
"""
@ -158,13 +177,14 @@ def post_test_rant() -> Optional[str]:
"plat": PLAT,
"guid": GUID,
"sid": SID,
"seid": SEID
"seid": SEID,
}
result = test_endpoint("POST", f"{BASE_URL}/devrant/rants", data=patch_auth(data))
if result["status_code"] == 200 and result.get("response", {}).get("success"):
return str(result["response"].get("rant_id", ""))
return None
# Post a test comment and return its id
def post_test_comment(rant_id: str) -> Optional[str]:
"""
@ -178,15 +198,19 @@ def post_test_comment(rant_id: str) -> Optional[str]:
"plat": PLAT,
"guid": GUID,
"sid": SID,
"seid": SEID
"seid": SEID,
}
result = test_endpoint("POST", f"{BASE_URL}/devrant/rants/{rant_id}/comments", data=patch_auth(data))
result = test_endpoint(
"POST", f"{BASE_URL}/devrant/rants/{rant_id}/comments", data=patch_auth(data)
)
if result["status_code"] == 200 and result.get("response", {}).get("success"):
return str(result["response"].get("comment_id", ""))
return None
# Test cases with docstrings
def test_register_user() -> None:
"""
Test user registration (valid and invalid).
@ -205,13 +229,14 @@ def test_register_user() -> None:
"plat": PLAT,
"guid": GUID,
"sid": SID,
"seid": SEID
"seid": SEID,
}
test_endpoint("POST", f"{BASE_URL}/users", data=patch_auth(params.copy()))
invalid_params = params.copy()
del invalid_params["email"]
test_endpoint("POST", f"{BASE_URL}/users", data=patch_auth(invalid_params))
def test_login_user() -> None:
"""
Test user login (valid and invalid). Already done in login_user(), but record here.
@ -229,10 +254,11 @@ def test_login_user() -> None:
"plat": PLAT,
"guid": GUID,
"sid": SID,
"seid": SEID
"seid": SEID,
}
test_endpoint("POST", f"{BASE_URL}/users/auth-token", data=patch_auth(params))
def test_edit_profile() -> None:
"""
Test editing user profile.
@ -249,10 +275,11 @@ def test_edit_profile() -> None:
"plat": PLAT,
"guid": GUID,
"sid": SID,
"seid": SEID
"seid": SEID,
}
test_endpoint("POST", f"{BASE_URL}/users/me/edit-profile", data=patch_auth(params))
def test_forgot_password() -> None:
"""
Test forgot password.
@ -265,10 +292,11 @@ def test_forgot_password() -> None:
"plat": PLAT,
"guid": GUID,
"sid": SID,
"seid": SEID
"seid": SEID,
}
test_endpoint("POST", f"{BASE_URL}/users/forgot-password", data=patch_auth(params))
def test_resend_confirm() -> None:
"""
Test resend confirmation email.
@ -276,13 +304,11 @@ def test_resend_confirm() -> None:
Payload: POST /users/me/resend-confirm with plat, guid, sid, seid, auth
Expected: success=true
"""
params: Dict[str, Any] = {
"plat": PLAT,
"guid": GUID,
"sid": SID,
"seid": SEID
}
test_endpoint("POST", f"{BASE_URL}/users/me/resend-confirm", data=patch_auth(params))
params: Dict[str, Any] = {"plat": PLAT, "guid": GUID, "sid": SID, "seid": SEID}
test_endpoint(
"POST", f"{BASE_URL}/users/me/resend-confirm", data=patch_auth(params)
)
def test_delete_account() -> None:
"""
@ -301,6 +327,7 @@ def test_delete_account() -> None:
# test_endpoint("DELETE", f"{BASE_URL}/users/me", params=patch_auth(params))
pass
def test_mark_news_read() -> None:
"""
Test mark news as read.
@ -313,9 +340,12 @@ def test_mark_news_read() -> None:
"plat": PLAT,
"guid": GUID,
"sid": SID,
"seid": SEID
"seid": SEID,
}
test_endpoint("POST", f"{BASE_URL}/users/me/mark-news-read", data=patch_auth(params))
test_endpoint(
"POST", f"{BASE_URL}/users/me/mark-news-read", data=patch_auth(params)
)
def test_get_rant() -> None:
"""
@ -330,9 +360,12 @@ def test_get_rant() -> None:
"plat": PLAT,
"guid": GUID,
"sid": SID,
"seid": SEID
"seid": SEID,
}
test_endpoint("GET", f"{BASE_URL}/devrant/rants/{TEST_RANT_ID}", params=patch_auth(params))
test_endpoint(
"GET", f"{BASE_URL}/devrant/rants/{TEST_RANT_ID}", params=patch_auth(params)
)
def test_post_rant() -> None:
"""
@ -344,6 +377,7 @@ def test_post_rant() -> None:
# Handled in setup
pass
def test_edit_rant() -> None:
"""
Test edit rant.
@ -351,11 +385,11 @@ def test_edit_rant() -> None:
Payload: POST /devrant/rants/{rant_id} with updated rant, tags, auth
Expected: success=true
"""
data: Dict[str, Any] = {
"rant": "Updated test rant",
"tags": "test,python,update"
}
test_endpoint("POST", f"{BASE_URL}/devrant/rants/{TEST_RANT_ID}", data=patch_auth(data))
data: Dict[str, Any] = {"rant": "Updated test rant", "tags": "test,python,update"}
test_endpoint(
"POST", f"{BASE_URL}/devrant/rants/{TEST_RANT_ID}", data=patch_auth(data)
)
def test_delete_rant() -> None:
"""
@ -364,13 +398,11 @@ def test_delete_rant() -> None:
Payload: DELETE /devrant/rants/{rant_id} with plat, guid, sid, seid, auth
Expected: success=true
"""
params: Dict[str, Any] = {
"plat": PLAT,
"guid": GUID,
"sid": SID,
"seid": SEID
}
test_endpoint("DELETE", f"{BASE_URL}/devrant/rants/{TEST_RANT_ID}", params=patch_auth(params))
params: Dict[str, Any] = {"plat": PLAT, "guid": GUID, "sid": SID, "seid": SEID}
test_endpoint(
"DELETE", f"{BASE_URL}/devrant/rants/{TEST_RANT_ID}", params=patch_auth(params)
)
def test_vote_rant() -> None:
"""
@ -384,12 +416,17 @@ def test_vote_rant() -> None:
"plat": PLAT,
"guid": GUID,
"sid": SID,
"seid": SEID
"seid": SEID,
}
test_endpoint("POST", f"{BASE_URL}/devrant/rants/{TEST_RANT_ID}/vote", data=patch_auth(params))
test_endpoint(
"POST", f"{BASE_URL}/devrant/rants/{TEST_RANT_ID}/vote", data=patch_auth(params)
)
params["vote"] = -1
params["reason"] = "1"
test_endpoint("POST", f"{BASE_URL}/devrant/rants/{TEST_RANT_ID}/vote", data=patch_auth(params))
test_endpoint(
"POST", f"{BASE_URL}/devrant/rants/{TEST_RANT_ID}/vote", data=patch_auth(params)
)
def test_favorite_rant() -> None:
"""
@ -398,14 +435,18 @@ def test_favorite_rant() -> None:
Payload: POST /devrant/rants/{rant_id}/favorite or /unfavorite with plat, guid, sid, seid, auth
Expected: success=true
"""
params: Dict[str, Any] = {
"plat": PLAT,
"guid": GUID,
"sid": SID,
"seid": SEID
}
test_endpoint("POST", f"{BASE_URL}/devrant/rants/{TEST_RANT_ID}/favorite", data=patch_auth(params))
test_endpoint("POST", f"{BASE_URL}/devrant/rants/{TEST_RANT_ID}/unfavorite", data=patch_auth(params))
params: Dict[str, Any] = {"plat": PLAT, "guid": GUID, "sid": SID, "seid": SEID}
test_endpoint(
"POST",
f"{BASE_URL}/devrant/rants/{TEST_RANT_ID}/favorite",
data=patch_auth(params),
)
test_endpoint(
"POST",
f"{BASE_URL}/devrant/rants/{TEST_RANT_ID}/unfavorite",
data=patch_auth(params),
)
def test_get_rant_feed() -> None:
"""
@ -414,14 +455,10 @@ def test_get_rant_feed() -> None:
Payload: GET /devrant/rants with plat, guid, sid, seid, auth
Expected: success=true, list of rants
"""
params: Dict[str, Any] = {
"plat": PLAT,
"guid": GUID,
"sid": SID,
"seid": SEID
}
params: Dict[str, Any] = {"plat": PLAT, "guid": GUID, "sid": SID, "seid": SEID}
test_endpoint("GET", f"{BASE_URL}/devrant/rants", params=patch_auth(params))
def test_get_comment() -> None:
"""
Test get single comment.
@ -434,9 +471,12 @@ def test_get_comment() -> None:
"plat": PLAT,
"guid": GUID,
"sid": SID,
"seid": SEID
"seid": SEID,
}
test_endpoint("GET", f"{BASE_URL}/comments/{TEST_COMMENT_ID}", params=patch_auth(params))
test_endpoint(
"GET", f"{BASE_URL}/comments/{TEST_COMMENT_ID}", params=patch_auth(params)
)
def test_post_comment() -> None:
"""
@ -448,6 +488,7 @@ def test_post_comment() -> None:
# Handled in setup
pass
def test_edit_comment() -> None:
"""
Test edit comment.
@ -455,10 +496,11 @@ def test_edit_comment() -> None:
Payload: POST /comments/{comment_id} with updated comment, auth
Expected: success=true
"""
data: Dict[str, Any] = {
"comment": "Updated test comment"
}
test_endpoint("POST", f"{BASE_URL}/comments/{TEST_COMMENT_ID}", data=patch_auth(data))
data: Dict[str, Any] = {"comment": "Updated test comment"}
test_endpoint(
"POST", f"{BASE_URL}/comments/{TEST_COMMENT_ID}", data=patch_auth(data)
)
def test_delete_comment() -> None:
"""
@ -467,13 +509,11 @@ def test_delete_comment() -> None:
Payload: DELETE /comments/{comment_id} with plat, guid, sid, seid, auth
Expected: success=true
"""
params: Dict[str, Any] = {
"plat": PLAT,
"guid": GUID,
"sid": SID,
"seid": SEID
}
test_endpoint("DELETE", f"{BASE_URL}/comments/{TEST_COMMENT_ID}", params=patch_auth(params))
params: Dict[str, Any] = {"plat": PLAT, "guid": GUID, "sid": SID, "seid": SEID}
test_endpoint(
"DELETE", f"{BASE_URL}/comments/{TEST_COMMENT_ID}", params=patch_auth(params)
)
def test_vote_comment() -> None:
"""
@ -487,9 +527,12 @@ def test_vote_comment() -> None:
"plat": PLAT,
"guid": GUID,
"sid": SID,
"seid": SEID
"seid": SEID,
}
test_endpoint("POST", f"{BASE_URL}/comments/{TEST_COMMENT_ID}/vote", data=patch_auth(params))
test_endpoint(
"POST", f"{BASE_URL}/comments/{TEST_COMMENT_ID}/vote", data=patch_auth(params)
)
def test_get_notif_feed() -> None:
"""
@ -504,10 +547,11 @@ def test_get_notif_feed() -> None:
"plat": PLAT,
"guid": GUID,
"sid": SID,
"seid": SEID
"seid": SEID,
}
test_endpoint("GET", f"{BASE_URL}/users/me/notif-feed", params=patch_auth(params))
def test_clear_notifications() -> None:
"""
Test clear notifications.
@ -515,13 +559,11 @@ def test_clear_notifications() -> None:
Payload: DELETE /users/me/notif-feed with plat, guid, sid, seid, auth
Expected: success=true
"""
params: Dict[str, Any] = {
"plat": PLAT,
"guid": GUID,
"sid": SID,
"seid": SEID
}
test_endpoint("DELETE", f"{BASE_URL}/users/me/notif-feed", params=patch_auth(params))
params: Dict[str, Any] = {"plat": PLAT, "guid": GUID, "sid": SID, "seid": SEID}
test_endpoint(
"DELETE", f"{BASE_URL}/users/me/notif-feed", params=patch_auth(params)
)
def test_beta_list_signup() -> None:
"""
@ -530,11 +572,10 @@ def test_beta_list_signup() -> None:
Payload: GET https://www.hexicallabs.com/api/beta-list with email, platform, app
Expected: Whatever the API returns (may not be JSON)
"""
params: Dict[str, Any] = {
"email": TEST_EMAIL,
"platform": "test_platform"
}
test_endpoint("GET", "https://www.hexicallabs.com/api/beta-list", params=patch_auth(params))
params: Dict[str, Any] = {"email": TEST_EMAIL, "platform": "test_platform"}
test_endpoint(
"GET", "https://www.hexicallabs.com/api/beta-list", params=patch_auth(params)
)
def main() -> None:
@ -570,5 +611,6 @@ def main() -> None:
test_delete_account()
save_results()
if __name__ == "__main__":
main()