grillkol/Stat-Dashboard
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add PostgreSQL analytics dashboard prototype and realistic sample data

Browse Source
This commit is contained in:
maxwes08
2026-11-03 09:13:00 +01:00
commit 70b94f8672
38 changed files with 2873 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

213
mockups/tests/test_database.py Normal file
View File

@@ -0,0 +1,213 @@
"""Integration checks use rolled-back fixture rows in the configured sample DB."""
from datetime import datetime, time, timedelta, timezone
from pathlib import Path
import sys
import unittest
import psycopg
sys.path.insert(0,str(Path(__file__).resolve().parents[1]))
from database import connect
from metrics_api import engagement, monetization, gamepasses
from sample_api import ApiError, dispatch, player_values, stage_funnel, player_detail, sessions
from psycopg.types.json import Jsonb
class MetricDefinitions(unittest.TestCase):
def setUp(self):
self.conn=connect()
self.game=self.conn.execute("INSERT INTO games(name) VALUES('Metric fixture') RETURNING id").fetchone()['id']
today=self.conn.execute('SELECT current_date AS day').fetchone()['day']
self.yesterday=datetime.combine(today-timedelta(days=1),time(1),timezone.utc)
self.today=datetime.combine(today,time(1),timezone.utc)
self.players=[]
for i in range(2):
self.players.append(self.conn.execute('INSERT INTO players(game_id,roblox_user_id,first_seen,last_seen) VALUES(%s,%s,%s,%s) RETURNING id',
(self.game,700+i,self.yesterday,self.today)).fetchone()['id'])
self.session_ids=[]
for pid,start,length in [(self.players[0],self.yesterday,300),(self.players[1],self.yesterday+timedelta(hours=1),120),(self.players[0],self.today,600)]:
self.session_ids.append(self.conn.execute('INSERT INTO sessions(game_id,player_id,started_at,ended_at) VALUES(%s,%s,%s,%s) RETURNING id',
(self.game,pid,start,start+timedelta(seconds=length))).fetchone()['id'])
self.products=[]
for i,price in enumerate((100,50)):
product=self.conn.execute('INSERT INTO products(game_id,roblox_product_id,name,current_price_robux) VALUES(%s,%s,%s,%s) RETURNING id',
(self.game,200+i,f'Product {i}',price)).fetchone()['id']
self.products.append(product)
self.conn.execute('INSERT INTO product_sales(game_id,product_id,player_id,session_id,receipt_id,purchased_at,unit_price_robux,quantity) VALUES(%s,%s,%s,%s,%s,%s,%s,%s)',
(self.game,product,self.players[i],self.session_ids[i],f'fixture-{i}',self.yesterday+timedelta(minutes=1),price,2 if i==0 else 1))
def tearDown(self):
self.conn.rollback()
self.conn.close()
def test_revenue_arppu_arpdau_and_decimal_product_counts(self):
data=monetization(self.conn,self.game,2)
m=data['metrics']
self.assertEqual(m['revenue'],250)
self.assertEqual(m['payingUsers'],2)
self.assertEqual(m['payerConversion'],100)
self.assertEqual(m['arppu'],125)
self.assertAlmostEqual(m['arpdau'],250/3)
self.assertEqual(m['productsPerPayer'],1.5)
by_product={p['id']:p for p in data['products']}
self.assertEqual(by_product[self.products[0]]['perPayer'],1)
self.assertEqual(by_product[self.products[1]]['perPayer'],.5)
self.assertEqual(sum(d['robux'] for d in data['daily']),250)
self.assertEqual(sum(h['robux'] for h in data['hourly']),250)
def test_rolling_mau_new_returning_and_five_minute_first_session(self):
m=engagement(self.conn,self.game,2)['metrics']
self.assertEqual(m['dau'],1)
self.assertEqual(m['mau'],2)
self.assertEqual(m['newUsersToday'],0)
self.assertEqual(m['returningUsersToday'],1)
self.assertEqual(m['firstSession5min'],50)
self.assertEqual(m['newSessionTime'],210)
self.assertEqual(m['returningSessionTime'],600)
self.assertEqual(m['totalPlaytime'],1020)
self.assertEqual(m['averagePlaytime'],510)
def test_gamepass_ownership_denominators_and_unknown_checks(self):
third=self.conn.execute('INSERT INTO players(game_id,roblox_user_id,first_seen) VALUES(%s,998,%s) RETURNING id',(self.game,self.today)).fetchone()['id']
gp=self.conn.execute("INSERT INTO gamepasses(game_id,roblox_gamepass_id,name) VALUES(%s,123,'VIP') RETURNING id",(self.game,)).fetchone()['id']
for pid,owned in [(self.players[0],True),(self.players[1],False),(third,True)]:
self.conn.execute('INSERT INTO gamepass_ownership(game_id,gamepass_id,player_id,is_owned,checked_at) VALUES(%s,%s,%s,%s,now())',(self.game,gp,pid,owned))
result=gamepasses(self.conn,self.game)['gamepasses'][0]
self.assertAlmostEqual(result['playerPercent'],200/3)
self.assertEqual(result['payerPercent'],50)
self.conn.execute('DELETE FROM gamepass_ownership WHERE game_id=%s AND player_id=%s',(self.game,third))
result=gamepasses(self.conn,self.game)['gamepasses'][0]
self.assertIsNone(result['playerPercent'])
self.assertEqual(result['payerPercent'],50)
self.conn.execute('DELETE FROM product_sales WHERE game_id=%s',(self.game,))
self.assertIsNone(gamepasses(self.conn,self.game)['gamepasses'][0]['payerPercent'])
def test_selected_player_session_aggregates(self):
m=engagement(self.conn,self.game,2,[self.players[1]])['metrics']
self.assertEqual(m['totalPlaytime'],120)
self.assertEqual(m['averagePlaytime'],120)
self.assertEqual(m['mau'],1)
self.assertEqual(m['dau'],0)
self.assertEqual(m['firstSession5min'],0)
self.assertEqual(engagement(self.conn,self.game,2,[])['metrics']['totalPlaytime'],0)
def test_two_history_values_use_actual_historical_sales(self):
pid=self.players[0]
self.conn.execute("INSERT INTO player_fields(game_id,key,label,format,source) VALUES(%s,'cash','Cash','number','snapshot')",(self.game,))
for at,data in [(self.yesterday+timedelta(seconds=30),{'cash':0,'customScore':17,'flag':True}),
(self.today+timedelta(minutes=12),{'cash':500,'customScore':99})]:
self.conn.execute('INSERT INTO player_snapshots(game_id,player_id,captured_at,data) VALUES(%s,%s,%s,%s)',(self.game,pid,at,Jsonb(data)))
detail=player_detail(self.conn,self.game,pid,{'value':['cash'],'value2':['robux']})
self.assertEqual([s['values'] for s in detail['series']],[[0,500],[0,200]])
fields={f['key'] for f in detail['historyFields']}
self.assertTrue({'robux','customScore','sessions','playtime','purchases','avgSession'}<=fields)
self.assertNotIn('flag',fields)
detail=player_detail(self.conn,self.game,pid,{'value':['playtime'],'value2':['sessions']})
self.assertEqual(detail['series'][0]['values'],[None,900])
self.assertEqual(detail['series'][1]['values'],[1,2])
with self.assertRaises(ApiError):
player_detail(self.conn,self.game,pid,{'value':['not_a_field']})
def test_join_session_only_for_recent_unended_sessions(self):
for age in [15,24*60]:
self.conn.execute("INSERT INTO sessions(game_id,player_id,started_at) VALUES(%s,%s,now()-%s*interval '1 minute')",(self.game,self.players[0],age))
records=sessions(self.conn,self.game,self.players[0])
self.assertEqual(sum(s['joinAvailable'] for s in records),1)
self.assertTrue(all(not s['joinAvailable'] for s in records if s['ended_at']))
def test_sale_price_is_independent_of_catalog_price(self):
self.conn.execute('UPDATE products SET current_price_robux=9999 WHERE game_id=%s',(self.game,))
self.assertEqual(monetization(self.conn,self.game,2)['metrics']['revenue'],250)
def test_unknown_session_end_is_not_a_failed_five_minute_session(self):
pid=self.conn.execute('INSERT INTO players(game_id,roblox_user_id,first_seen) VALUES(%s,999,%s) RETURNING id',(self.game,self.today)).fetchone()['id']
self.conn.execute('INSERT INTO sessions(game_id,player_id,started_at) VALUES(%s,%s,%s)',(self.game,pid,self.today))
self.assertEqual(engagement(self.conn,self.game,2)['metrics']['firstSession5min'],50)
row=self.conn.execute('SELECT * FROM player_statistics WHERE game_id=%s AND id=%s',(self.game,pid)).fetchone()
player=player_values(row)
self.assertIsNone(player['avgSession'])
self.assertIsNone(player['playtime'])
self.assertIsNone(player.get('level'))
self.assertEqual(player['robux'],0)
def test_cross_game_player_reference_is_rejected(self):
other=self.conn.execute("INSERT INTO games(name) VALUES('Other fixture') RETURNING id").fetchone()['id']
with self.assertRaises(psycopg.errors.ForeignKeyViolation):
with self.conn.transaction():
self.conn.execute('INSERT INTO sessions(game_id,player_id,started_at) VALUES(%s,%s,%s)',(other,self.players[0],self.today))
def test_duplicate_receipt_is_rejected(self):
with self.assertRaises(psycopg.errors.UniqueViolation):
with self.conn.transaction():
self.conn.execute('INSERT INTO product_sales(game_id,product_id,receipt_id,purchased_at,unit_price_robux) VALUES(%s,%s,%s,%s,100)',
(self.game,self.products[0],'fixture-0',self.today))
class SeededApi(unittest.TestCase):
@classmethod
def setUpClass(cls):
cls.games=dispatch('/api/games',{})['games']
cls.game=cls.games[0]['id']
def test_daily_event_lines_match_summary(self):
data=dispatch(f'/api/games/{self.game}/events',{'days':['7']})
self.assertNotIn('events',data)
self.assertEqual(len(data['labels']),7)
counts={s['name']:sum(s['values']) for s in data['series']}
self.assertEqual(counts,{r['name']:r['total'] for r in data['eventSummary']})
self.assertEqual(sum(counts.values()),data['metrics']['events'])
def test_session_groups_filter_counts_and_chart(self):
all_data=dispatch(f'/api/games/{self.game}/sessions',{})
for segment in ['whales','playtime','first']:
data=dispatch(f'/api/games/{self.game}/sessions',{'segment':[segment]})
self.assertLessEqual(data['metrics']['sessions'],all_data['metrics']['sessions'])
self.assertNotIn('sessions',data)
line=next(s for s in data['series'] if s['name']=='Sessions')
self.assertEqual(sum(line['values']),data['metrics']['sessions'])
def test_internal_game_root_and_dataset_scope(self):
for game in self.games:
root=dispatch(f"/api/games/{game['id']}",{})
self.assertEqual(root['gameId'],game['id'])
self.assertIn('robloxExperienceId',root['game'])
self.assertTrue(all(f"/games/{game['id']}/" in route for route in root['datasets'].values()))
players=dispatch(f"/api/games/{game['id']}/players",{})
self.assertTrue(all(p['gameId']==game['id'] for p in players['playerRows']))
def test_real_player_detail_and_cross_game_lookup(self):
player=dispatch(f'/api/games/{self.game}/players',{})['playerRows'][0]
detail=dispatch(f"/api/games/{self.game}/players/{player['playerId']}",{})
self.assertEqual(detail['player']['playerId'],player['playerId'])
self.assertTrue(all(s['player_id']==player['playerId'] for s in detail['sessions']))
with self.assertRaises(ApiError) as error:
dispatch(f"/api/games/{self.games[1]['id']}/players/{player['playerId']}",{})
self.assertEqual(error.exception.status,404)
def test_missing_stats_sort_last_and_zero_is_preserved(self):
page=dispatch(f'/api/games/{self.game}/players',{'sort':['cash'],'direction':['asc'],'page':['999999']})
self.assertIsNone(page['playerRows'][-1].get('cash'))
page=dispatch(f'/api/games/{self.game}/players',{'sort':['robux'],'direction':['asc']})
self.assertEqual(page['playerRows'][0]['robux'],0)
def test_game_level_categories_and_missing_root_property(self):
with connect() as conn:
self.assertGreater(conn.execute('SELECT count(*) n FROM events WHERE game_id=%s AND player_id IS NULL',(self.game,)).fetchone()['n'],0)
self.assertGreater(conn.execute('SELECT count(*) n FROM game_metrics WHERE game_id=%s',(self.game,)).fetchone()['n'],0)
self.assertTrue(any(g['robloxExperienceId'] is None for g in self.games))
class ActivityStages(unittest.TestCase):
def test_ordered_reach_and_dropoff(self):
definition={'start_event':'start','end_event':'end','configuration':{'steps':['prompt','purchase']}}
runs=[{'stage_progress':p} for p in [['start','prompt','purchase','end'],['start','prompt'],['start'],['start','purchase','prompt','end']]]
stages=stage_funnel(runs,definition)
self.assertEqual([s['runs'] for s in stages],[4,3,1,1])
self.assertEqual([s['reachPercent'] for s in stages],[100,75,25,25])
self.assertEqual(stages[1]['dropoffPercent'],25)
self.assertAlmostEqual(stages[2]['dropoffPercent'],200/3)
runs.append({'stage_progress':None})
self.assertIsNone(stage_funnel(runs,definition)[1]['dropoffPercent'])
self.assertIsNone(stage_funnel([],definition)[0]['reachPercent'])
if __name__=='__main__': unittest.main()