2019-12-08 22:43:49 +01:00
import random
import time
from typing import Dict , List , Optional
from bemani . common import Time
from bemani . client . base import BaseClient
from bemani . protocol import Node
class ReflecBeatLimelight ( BaseClient ) :
NAME = ' T E S T '
def verify_log_pcb_status ( self , loc : str ) - > None :
call = self . call_node ( )
pcb = Node . void ( ' log ' )
pcb . set_attribute ( ' method ' , ' pcb_status ' )
pcb . add_child ( Node . string ( ' lid ' , loc ) )
pcb . add_child ( Node . s32 ( ' cnt ' , 0 ) )
call . add_child ( pcb )
# Swap with server
resp = self . exchange ( ' ' , call )
# Verify that response is correct
self . assert_path ( resp , " response/log/@status " )
def verify_pcbinfo_get ( self , loc : str ) - > None :
call = self . call_node ( )
pcb = Node . void ( ' pcbinfo ' )
pcb . set_attribute ( ' method ' , ' get ' )
pcb . add_child ( Node . string ( ' lid ' , loc ) )
call . add_child ( pcb )
# Swap with server
resp = self . exchange ( ' ' , call )
# Verify that response is correct
self . assert_path ( resp , " response/pcbinfo/info/name " )
self . assert_path ( resp , " response/pcbinfo/info/pref " )
self . assert_path ( resp , " response/pcbinfo/info/close " )
self . assert_path ( resp , " response/pcbinfo/info/hour " )
self . assert_path ( resp , " response/pcbinfo/info/min " )
def verify_sysinfo_get ( self ) - > None :
call = self . call_node ( )
info = Node . void ( ' sysinfo ' )
info . set_attribute ( ' method ' , ' get ' )
call . add_child ( info )
# Swap with server
resp = self . exchange ( ' ' , call )
# Verify that response is correct
self . assert_path ( resp , " response/sysinfo/trd " )
def verify_ranking_read ( self ) - > None :
call = self . call_node ( )
ranking = Node . void ( ' ranking ' )
ranking . set_attribute ( ' method ' , ' read ' )
call . add_child ( ranking )
# Swap with server
resp = self . exchange ( ' ' , call )
# Verify that response is correct
self . assert_path ( resp , " response/ranking/lic_10/time " )
self . assert_path ( resp , " response/ranking/org_10/time " )
def verify_player_start ( self , refid : str ) - > None :
call = self . call_node ( )
player = Node . void ( ' player ' )
player . set_attribute ( ' method ' , ' start ' )
player . add_child ( Node . string ( ' rid ' , refid ) )
player . add_child ( Node . s32 ( ' ver ' , 0 ) )
call . add_child ( player )
# Swap with server
resp = self . exchange ( ' ' , call )
# Verify that response is correct
self . assert_path ( resp , " response/player/is_suc " )
self . assert_path ( resp , " response/player/unlock_music " )
self . assert_path ( resp , " response/player/unlock_item " )
self . assert_path ( resp , " response/player/item_lock_ctrl " )
self . assert_path ( resp , " response/player/lincle_link_4/qpro " )
self . assert_path ( resp , " response/player/lincle_link_4/glass " )
self . assert_path ( resp , " response/player/lincle_link_4/treasure " )
self . assert_path ( resp , " response/player/lincle_link_4/for_iidx_0_0 " )
self . assert_path ( resp , " response/player/lincle_link_4/for_iidx_0_1 " )
self . assert_path ( resp , " response/player/lincle_link_4/for_iidx_0_2 " )
self . assert_path ( resp , " response/player/lincle_link_4/for_iidx_0_3 " )
self . assert_path ( resp , " response/player/lincle_link_4/for_iidx_0_4 " )
self . assert_path ( resp , " response/player/lincle_link_4/for_iidx_0_5 " )
self . assert_path ( resp , " response/player/lincle_link_4/for_iidx_0_6 " )
self . assert_path ( resp , " response/player/lincle_link_4/for_iidx_0 " )
self . assert_path ( resp , " response/player/lincle_link_4/for_iidx_1 " )
self . assert_path ( resp , " response/player/lincle_link_4/for_iidx_2 " )
self . assert_path ( resp , " response/player/lincle_link_4/for_iidx_3 " )
self . assert_path ( resp , " response/player/lincle_link_4/for_iidx_4 " )
self . assert_path ( resp , " response/player/lincle_link_4/for_rb_0_0 " )
self . assert_path ( resp , " response/player/lincle_link_4/for_rb_0_1 " )
self . assert_path ( resp , " response/player/lincle_link_4/for_rb_0_2 " )
self . assert_path ( resp , " response/player/lincle_link_4/for_rb_0_3 " )
self . assert_path ( resp , " response/player/lincle_link_4/for_rb_0_4 " )
self . assert_path ( resp , " response/player/lincle_link_4/for_rb_0_5 " )
self . assert_path ( resp , " response/player/lincle_link_4/for_rb_0_6 " )
self . assert_path ( resp , " response/player/lincle_link_4/for_rb_0 " )
self . assert_path ( resp , " response/player/lincle_link_4/for_rb_1 " )
self . assert_path ( resp , " response/player/lincle_link_4/for_rb_2 " )
self . assert_path ( resp , " response/player/lincle_link_4/for_rb_3 " )
self . assert_path ( resp , " response/player/lincle_link_4/for_rb_4 " )
self . assert_path ( resp , " response/player/lincle_link_4/qproflg " )
self . assert_path ( resp , " response/player/lincle_link_4/glassflg " )
self . assert_path ( resp , " response/player/lincle_link_4/complete " )
def verify_player_delete ( self , refid : str ) - > None :
call = self . call_node ( )
player = Node . void ( ' player ' )
player . set_attribute ( ' method ' , ' delete ' )
player . add_child ( Node . string ( ' rid ' , refid ) )
call . add_child ( player )
# Swap with server
resp = self . exchange ( ' ' , call )
# Verify that response is correct
self . assert_path ( resp , " response/player/@status " )
def verify_player_end ( self , refid : str ) - > None :
call = self . call_node ( )
player = Node . void ( ' player ' )
player . set_attribute ( ' method ' , ' end ' )
player . add_child ( Node . string ( ' rid ' , refid ) )
player . add_child ( Node . s32 ( ' status ' , 4 ) )
call . add_child ( player )
# Swap with server
resp = self . exchange ( ' ' , call )
# Verify that response is correct
self . assert_path ( resp , " response/player " )
def verify_player_read ( self , refid : str , location : str ) - > List [ Dict [ str , int ] ] :
call = self . call_node ( )
player = Node . void ( ' player ' )
player . set_attribute ( ' method ' , ' read ' )
player . add_child ( Node . string ( ' rid ' , refid ) )
player . add_child ( Node . string ( ' lid ' , location ) )
player . add_child ( Node . s32 ( ' ver ' , 2 ) )
call . add_child ( player )
# Swap with server
resp = self . exchange ( ' ' , call )
# Verify that response is correct
self . assert_path ( resp , " response/player/pdata/base/uid " )
self . assert_path ( resp , " response/player/pdata/base/name " )
self . assert_path ( resp , " response/player/pdata/base/icon_id " )
self . assert_path ( resp , " response/player/pdata/base/lv " )
self . assert_path ( resp , " response/player/pdata/base/exp " )
self . assert_path ( resp , " response/player/pdata/base/mg " )
self . assert_path ( resp , " response/player/pdata/base/ap " )
self . assert_path ( resp , " response/player/pdata/base/pc " )
self . assert_path ( resp , " response/player/pdata/base/uattr " )
self . assert_path ( resp , " response/player/pdata/con/day " )
self . assert_path ( resp , " response/player/pdata/con/cnt " )
self . assert_path ( resp , " response/player/pdata/con/total_cnt " )
self . assert_path ( resp , " response/player/pdata/con/last " )
self . assert_path ( resp , " response/player/pdata/con/now " )
self . assert_path ( resp , " response/player/pdata/team/id " )
self . assert_path ( resp , " response/player/pdata/team/name " )
self . assert_path ( resp , " response/player/pdata/custom/s_gls " )
self . assert_path ( resp , " response/player/pdata/custom/bgm_m " )
self . assert_path ( resp , " response/player/pdata/custom/st_f " )
self . assert_path ( resp , " response/player/pdata/custom/st_bg " )
self . assert_path ( resp , " response/player/pdata/custom/st_bg_b " )
self . assert_path ( resp , " response/player/pdata/custom/eff_e " )
self . assert_path ( resp , " response/player/pdata/custom/se_s " )
self . assert_path ( resp , " response/player/pdata/custom/se_s_v " )
self . assert_path ( resp , " response/player/pdata/custom/last_music_id " )
self . assert_path ( resp , " response/player/pdata/custom/last_note_grade " )
self . assert_path ( resp , " response/player/pdata/custom/sort_type " )
self . assert_path ( resp , " response/player/pdata/custom/narrowdown_type " )
self . assert_path ( resp , " response/player/pdata/custom/is_begginer " )
self . assert_path ( resp , " response/player/pdata/custom/is_tut " )
self . assert_path ( resp , " response/player/pdata/custom/symbol_chat_0 " )
self . assert_path ( resp , " response/player/pdata/custom/symbol_chat_1 " )
self . assert_path ( resp , " response/player/pdata/custom/gauge_style " )
self . assert_path ( resp , " response/player/pdata/custom/obj_shade " )
self . assert_path ( resp , " response/player/pdata/custom/obj_size " )
self . assert_path ( resp , " response/player/pdata/custom/byword " )
self . assert_path ( resp , " response/player/pdata/custom/is_auto_byword " )
self . assert_path ( resp , " response/player/pdata/custom/is_tweet " )
self . assert_path ( resp , " response/player/pdata/custom/is_link_twitter " )
self . assert_path ( resp , " response/player/pdata/custom/mrec_type " )
self . assert_path ( resp , " response/player/pdata/custom/card_disp_type " )
self . assert_path ( resp , " response/player/pdata/custom/tab_sel " )
self . assert_path ( resp , " response/player/pdata/custom/hidden_param " )
self . assert_path ( resp , " response/player/pdata/released " )
self . assert_path ( resp , " response/player/pdata/record " )
self . assert_path ( resp , " response/player/pdata/cmnt " )
self . assert_path ( resp , " response/player/pdata/rival " )
self . assert_path ( resp , " response/player/pdata/glass " )
self . assert_path ( resp , " response/player/pdata/fav_music_slot " )
self . assert_path ( resp , " response/player/pdata/narrow_down/adv_param " )
if resp . child_value ( ' player/pdata/base/name ' ) != self . NAME :
2020-01-07 22:29:07 +01:00
raise Exception ( f ' Invalid name { resp . child_value ( " player/pdata/base/name " ) } returned on profile read! ' )
2019-12-08 22:43:49 +01:00
scores = [ ]
for child in resp . child ( ' player/pdata/record ' ) . children :
if child . name != ' rec ' :
continue
score = {
' id ' : child . child_value ( ' mid ' ) ,
' chart ' : child . child_value ( ' ng ' ) ,
' clear_type ' : child . child_value ( ' mrec_0/ct ' ) ,
' achievement_rate ' : child . child_value ( ' mrec_0/ar ' ) ,
' score ' : child . child_value ( ' mrec_0/bs ' ) ,
' combo ' : child . child_value ( ' mrec_0/mc ' ) ,
' miss_count ' : child . child_value ( ' mrec_0/bmc ' ) ,
}
scores . append ( score )
return scores
def verify_player_write ( self , refid : str , extid : int , loc : str , records : List [ Dict [ str , int ] ] , scores : List [ Dict [ str , int ] ] ) - > int :
call = self . call_node ( )
player = Node . void ( ' player ' )
call . add_child ( player )
player . set_attribute ( ' method ' , ' write ' )
player . add_child ( Node . string ( ' rid ' , refid ) )
player . add_child ( Node . string ( ' lid ' , loc ) )
player . add_child ( Node . u64 ( ' begin_time ' , Time . now ( ) * 1000 ) )
player . add_child ( Node . u64 ( ' end_time ' , Time . now ( ) * 1000 ) )
pdata = Node . void ( ' pdata ' )
player . add_child ( pdata )
base = Node . void ( ' base ' )
pdata . add_child ( base )
base . add_child ( Node . s32 ( ' uid ' , extid ) )
base . add_child ( Node . string ( ' name ' , self . NAME ) )
base . add_child ( Node . s16 ( ' icon_id ' , 0 ) )
base . add_child ( Node . s16 ( ' lv ' , 1 ) )
base . add_child ( Node . s32 ( ' exp ' , 0 ) )
base . add_child ( Node . s16 ( ' mg ' , 0 ) )
base . add_child ( Node . s16 ( ' ap ' , 0 ) )
base . add_child ( Node . s32 ( ' pc ' , 0 ) )
base . add_child ( Node . s32 ( ' uattr ' , 0 ) )
con = Node . void ( ' con ' )
pdata . add_child ( con )
con . add_child ( Node . s32 ( ' day ' , 0 ) )
con . add_child ( Node . s32 ( ' cnt ' , 0 ) )
con . add_child ( Node . s32 ( ' total_cnt ' , 0 ) )
con . add_child ( Node . s32 ( ' last ' , 0 ) )
con . add_child ( Node . s32 ( ' now ' , 0 ) )
custom = Node . void ( ' custom ' )
pdata . add_child ( custom )
custom . add_child ( Node . u8 ( ' s_gls ' , 0 ) )
custom . add_child ( Node . u8 ( ' bgm_m ' , 0 ) )
custom . add_child ( Node . u8 ( ' st_f ' , 0 ) )
custom . add_child ( Node . u8 ( ' st_bg ' , 0 ) )
custom . add_child ( Node . u8 ( ' st_bg_b ' , 100 ) )
custom . add_child ( Node . u8 ( ' eff_e ' , 0 ) )
custom . add_child ( Node . u8 ( ' se_s ' , 0 ) )
custom . add_child ( Node . u8 ( ' se_s_v ' , 100 ) )
custom . add_child ( Node . s16 ( ' last_music_id ' , 85 ) )
custom . add_child ( Node . u8 ( ' last_note_grade ' , 0 ) )
custom . add_child ( Node . u8 ( ' sort_type ' , 0 ) )
custom . add_child ( Node . u8 ( ' narrowdown_type ' , 0 ) )
custom . add_child ( Node . bool ( ' is_begginer ' , False ) )
custom . add_child ( Node . bool ( ' is_tut ' , False ) )
custom . add_child ( Node . s16_array ( ' symbol_chat_0 ' , [ 0 , 1 , 2 , 3 , 4 , 5 ] ) )
custom . add_child ( Node . s16_array ( ' symbol_chat_1 ' , [ 0 , 1 , 2 , 3 , 4 , 5 ] ) )
custom . add_child ( Node . u8 ( ' gauge_style ' , 0 ) )
custom . add_child ( Node . u8 ( ' obj_shade ' , 0 ) )
custom . add_child ( Node . u8 ( ' obj_size ' , 0 ) )
custom . add_child ( Node . s16_array ( ' byword ' , [ 0 , 0 ] ) )
custom . add_child ( Node . bool_array ( ' is_auto_byword ' , [ True , True ] ) )
custom . add_child ( Node . bool ( ' is_tweet ' , False ) )
custom . add_child ( Node . bool ( ' is_link_twitter ' , False ) )
custom . add_child ( Node . s16 ( ' mrec_type ' , 0 ) )
custom . add_child ( Node . s16 ( ' card_disp_type ' , 0 ) )
custom . add_child ( Node . s16 ( ' tab_sel ' , 0 ) )
custom . add_child ( Node . s32_array ( ' hidden_param ' , [ 0 , 0 , 0 , 0 , 0 , 0 , 0 , 0 , 0 , 0 , 0 , 0 , 1 , 0 , 0 , 1 , 0 , 0 , 0 , 0 ] ) )
pdata . add_child ( Node . void ( ' released ' ) )
pdata . add_child ( Node . void ( ' rival ' ) )
pdata . add_child ( Node . void ( ' glass ' ) )
pdata . add_child ( Node . void ( ' fav_music_slot ' ) )
lincle_link_4 = Node . void ( ' lincle_link_4 ' )
pdata . add_child ( lincle_link_4 )
lincle_link_4 . add_child ( Node . u32 ( ' qpro_add ' , 0 ) )
lincle_link_4 . add_child ( Node . u32 ( ' glass_add ' , 0 ) )
lincle_link_4 . add_child ( Node . bool ( ' for_iidx_0_0 ' , False ) )
lincle_link_4 . add_child ( Node . bool ( ' for_iidx_0_1 ' , False ) )
lincle_link_4 . add_child ( Node . bool ( ' for_iidx_0_2 ' , False ) )
lincle_link_4 . add_child ( Node . bool ( ' for_iidx_0_3 ' , False ) )
lincle_link_4 . add_child ( Node . bool ( ' for_iidx_0_4 ' , False ) )
lincle_link_4 . add_child ( Node . bool ( ' for_iidx_0_5 ' , False ) )
lincle_link_4 . add_child ( Node . bool ( ' for_iidx_0_6 ' , False ) )
lincle_link_4 . add_child ( Node . bool ( ' for_iidx_0 ' , False ) )
lincle_link_4 . add_child ( Node . bool ( ' for_iidx_1 ' , False ) )
lincle_link_4 . add_child ( Node . bool ( ' for_iidx_2 ' , False ) )
lincle_link_4 . add_child ( Node . bool ( ' for_iidx_3 ' , False ) )
lincle_link_4 . add_child ( Node . bool ( ' for_iidx_4 ' , False ) )
lincle_link_4 . add_child ( Node . bool ( ' for_rb_0_0 ' , False ) )
lincle_link_4 . add_child ( Node . bool ( ' for_rb_0_1 ' , False ) )
lincle_link_4 . add_child ( Node . bool ( ' for_rb_0_2 ' , False ) )
lincle_link_4 . add_child ( Node . bool ( ' for_rb_0_3 ' , False ) )
lincle_link_4 . add_child ( Node . bool ( ' for_rb_0_4 ' , False ) )
lincle_link_4 . add_child ( Node . bool ( ' for_rb_0_5 ' , False ) )
lincle_link_4 . add_child ( Node . bool ( ' for_rb_0_6 ' , False ) )
lincle_link_4 . add_child ( Node . bool ( ' for_rb_0 ' , False ) )
lincle_link_4 . add_child ( Node . bool ( ' for_rb_1 ' , False ) )
lincle_link_4 . add_child ( Node . bool ( ' for_rb_2 ' , False ) )
lincle_link_4 . add_child ( Node . bool ( ' for_rb_3 ' , False ) )
lincle_link_4 . add_child ( Node . bool ( ' for_rb_4 ' , False ) )
# First, filter down to only records that are also in the battle log
def key ( thing : Dict [ str , int ] ) - > str :
2020-01-07 22:29:07 +01:00
return f ' { thing [ " id " ] } - { thing [ " chart " ] } '
2019-12-08 22:43:49 +01:00
updates = [ key ( score ) for score in scores ]
sortedrecords = { key ( record ) : record for record in records if key ( record ) in updates }
# Now, see what records need updating and update them
for score in scores :
if key ( score ) in sortedrecords :
# Had a record, need to merge
record = sortedrecords [ key ( score ) ]
else :
# First time playing
record = {
' clear_type ' : 0 ,
' achievement_rate ' : 0 ,
' score ' : 0 ,
' combo ' : 0 ,
' miss_count ' : 999999999 ,
}
sortedrecords [ key ( score ) ] = {
' id ' : score [ ' id ' ] ,
' chart ' : score [ ' chart ' ] ,
' clear_type ' : max ( record [ ' clear_type ' ] , score [ ' clear_type ' ] ) ,
' achievement_rate ' : max ( record [ ' achievement_rate ' ] , score [ ' achievement_rate ' ] ) ,
' score ' : max ( record [ ' score ' ] , score [ ' score ' ] ) ,
' combo ' : max ( record [ ' combo ' ] , score [ ' combo ' ] ) ,
' miss_count ' : min ( record [ ' miss_count ' ] , score [ ' miss_count ' ] ) ,
}
# Finally, send the records and battle logs
recordnode = Node . void ( ' record ' )
pdata . add_child ( recordnode )
blog = Node . void ( ' blog ' )
pdata . add_child ( blog )
for ( _ , record ) in sortedrecords . items ( ) :
rec = Node . void ( ' rec ' )
recordnode . add_child ( rec )
rec . add_child ( Node . u16 ( ' mid ' , record [ ' id ' ] ) )
rec . add_child ( Node . u8 ( ' ng ' , record [ ' chart ' ] ) )
rec . add_child ( Node . s32 ( ' point ' , 2 ) )
rec . add_child ( Node . s32 ( ' played_time ' , Time . now ( ) ) )
mrec_0 = Node . void ( ' mrec_0 ' )
rec . add_child ( mrec_0 )
mrec_0 . add_child ( Node . s32 ( ' win ' , 1 ) )
mrec_0 . add_child ( Node . s32 ( ' lose ' , 0 ) )
mrec_0 . add_child ( Node . s32 ( ' draw ' , 0 ) )
mrec_0 . add_child ( Node . u8 ( ' ct ' , record [ ' clear_type ' ] ) )
mrec_0 . add_child ( Node . s16 ( ' ar ' , record [ ' achievement_rate ' ] ) )
mrec_0 . add_child ( Node . s32 ( ' bs ' , record [ ' score ' ] ) )
mrec_0 . add_child ( Node . s16 ( ' mc ' , record [ ' combo ' ] ) )
mrec_0 . add_child ( Node . s16 ( ' bmc ' , record [ ' miss_count ' ] ) )
mrec_1 = Node . void ( ' mrec_1 ' )
rec . add_child ( mrec_1 )
mrec_1 . add_child ( Node . s32 ( ' win ' , 0 ) )
mrec_1 . add_child ( Node . s32 ( ' lose ' , 0 ) )
mrec_1 . add_child ( Node . s32 ( ' draw ' , 0 ) )
mrec_1 . add_child ( Node . u8 ( ' ct ' , 0 ) )
mrec_1 . add_child ( Node . s16 ( ' ar ' , 0 ) )
mrec_1 . add_child ( Node . s32 ( ' bs ' , 0 ) )
mrec_1 . add_child ( Node . s16 ( ' mc ' , 0 ) )
mrec_1 . add_child ( Node . s16 ( ' bmc ' , - 1 ) )
scoreid = 0
for score in scores :
log = Node . void ( ' log ' )
blog . add_child ( log )
log . add_child ( Node . u8 ( ' id ' , scoreid ) )
log . add_child ( Node . u16 ( ' mid ' , score [ ' id ' ] ) )
log . add_child ( Node . u8 ( ' ng ' , score [ ' chart ' ] ) )
log . add_child ( Node . u8 ( ' mt ' , 0 ) )
log . add_child ( Node . u8 ( ' rt ' , 0 ) )
log . add_child ( Node . s32 ( ' ruid ' , 0 ) )
myself = Node . void ( ' myself ' )
log . add_child ( myself )
myself . add_child ( Node . s16 ( ' mg ' , 0 ) )
myself . add_child ( Node . s16 ( ' ap ' , 0 ) )
myself . add_child ( Node . u8 ( ' ct ' , score [ ' clear_type ' ] ) )
myself . add_child ( Node . s32 ( ' s ' , score [ ' score ' ] ) )
myself . add_child ( Node . s16 ( ' ar ' , score [ ' achievement_rate ' ] ) )
rival = Node . void ( ' rival ' )
log . add_child ( rival )
rival . add_child ( Node . s16 ( ' mg ' , 0 ) )
rival . add_child ( Node . s16 ( ' ap ' , 0 ) )
rival . add_child ( Node . u8 ( ' ct ' , 2 ) )
rival . add_child ( Node . s32 ( ' s ' , 177 ) )
rival . add_child ( Node . s16 ( ' ar ' , 500 ) )
log . add_child ( Node . s32 ( ' time ' , Time . now ( ) ) )
scoreid = scoreid + 1
# Swap with server
resp = self . exchange ( ' ' , call )
# Verify that response is correct
self . assert_path ( resp , " response/player/uid " )
self . assert_path ( resp , " response/player/time " )
return resp . child_value ( ' player/uid ' )
def verify_log_play ( self , extid : int , loc : str , scores : List [ Dict [ str , int ] ] ) - > None :
call = self . call_node ( )
log = Node . void ( ' log ' )
call . add_child ( log )
log . set_attribute ( ' method ' , ' play ' )
log . add_child ( Node . s32 ( ' uid ' , extid ) )
log . add_child ( Node . string ( ' lid ' , loc ) )
play = Node . void ( ' play ' )
log . add_child ( play )
play . add_child ( Node . s16 ( ' stage ' , len ( scores ) ) )
play . add_child ( Node . s32 ( ' sec ' , 700 ) )
scoreid = 0
for score in scores :
rec = Node . void ( ' rec ' )
log . add_child ( rec )
rec . add_child ( Node . s16 ( ' idx ' , scoreid ) )
rec . add_child ( Node . s16 ( ' mid ' , score [ ' id ' ] ) )
rec . add_child ( Node . s16 ( ' grade ' , score [ ' chart ' ] ) )
rec . add_child ( Node . s16 ( ' color ' , 0 ) )
rec . add_child ( Node . s16 ( ' match ' , 0 ) )
rec . add_child ( Node . s16 ( ' res ' , 0 ) )
rec . add_child ( Node . s32 ( ' score ' , score [ ' score ' ] ) )
rec . add_child ( Node . s16 ( ' mc ' , score [ ' combo ' ] ) )
rec . add_child ( Node . s16 ( ' jt_jr ' , 0 ) )
rec . add_child ( Node . s16 ( ' jt_ju ' , 0 ) )
rec . add_child ( Node . s16 ( ' jt_gr ' , 0 ) )
rec . add_child ( Node . s16 ( ' jt_gd ' , 0 ) )
rec . add_child ( Node . s16 ( ' jt_ms ' , score [ ' miss_count ' ] ) )
rec . add_child ( Node . s32 ( ' sec ' , 200 ) )
scoreid = scoreid + 1
# Swap with server
resp = self . exchange ( ' ' , call )
# Verify that response is correct
self . assert_path ( resp , " response/log/@status " )
def verify_lobby_read ( self , location : str , extid : int ) - > None :
call = self . call_node ( )
lobby = Node . void ( ' lobby ' )
lobby . set_attribute ( ' method ' , ' read ' )
lobby . add_child ( Node . s32 ( ' uid ' , extid ) )
lobby . add_child ( Node . u8 ( ' m_grade ' , 255 ) )
lobby . add_child ( Node . string ( ' lid ' , location ) )
lobby . add_child ( Node . s32 ( ' max ' , 128 ) )
lobby . add_child ( Node . s32_array ( ' friend ' , [ ] ) )
call . add_child ( lobby )
# Swap with server
resp = self . exchange ( ' ' , call )
# Verify that response is correct
self . assert_path ( resp , " response/lobby/interval " )
self . assert_path ( resp , " response/lobby/interval_p " )
def verify_lobby_entry ( self , location : str , extid : int ) - > int :
call = self . call_node ( )
lobby = Node . void ( ' lobby ' )
lobby . set_attribute ( ' method ' , ' entry ' )
e = Node . void ( ' e ' )
lobby . add_child ( e )
e . add_child ( Node . s32 ( ' eid ' , 0 ) )
e . add_child ( Node . u16 ( ' mid ' , 79 ) )
e . add_child ( Node . u8 ( ' ng ' , 0 ) )
e . add_child ( Node . s32 ( ' uid ' , extid ) )
e . add_child ( Node . s32 ( ' uattr ' , 0 ) )
e . add_child ( Node . string ( ' pn ' , self . NAME ) )
e . add_child ( Node . s16 ( ' mg ' , 0 ) )
e . add_child ( Node . s32 ( ' mopt ' , 0 ) )
e . add_child ( Node . s32 ( ' tid ' , 0 ) )
e . add_child ( Node . string ( ' tn ' , ' ' ) )
e . add_child ( Node . s32 ( ' topt ' , 0 ) )
e . add_child ( Node . string ( ' lid ' , location ) )
e . add_child ( Node . string ( ' sn ' , ' ' ) )
e . add_child ( Node . u8 ( ' pref ' , 51 ) )
e . add_child ( Node . s8 ( ' stg ' , 0 ) )
e . add_child ( Node . s8 ( ' pside ' , 0 ) )
e . add_child ( Node . s16 ( ' eatime ' , 30 ) )
e . add_child ( Node . u8_array ( ' ga ' , [ 127 , 0 , 0 , 1 ] ) )
e . add_child ( Node . u16 ( ' gp ' , 10007 ) )
e . add_child ( Node . u8_array ( ' la ' , [ 16 , 0 , 0 , 0 ] ) )
lobby . add_child ( Node . s32_array ( ' friend ' , [ ] ) )
call . add_child ( lobby )
# Swap with server
resp = self . exchange ( ' ' , call )
# Verify that response is correct
self . assert_path ( resp , " response/lobby/eid " )
self . assert_path ( resp , " response/lobby/interval " )
self . assert_path ( resp , " response/lobby/interval_p " )
self . assert_path ( resp , " response/lobby/e/eid " )
self . assert_path ( resp , " response/lobby/e/mid " )
self . assert_path ( resp , " response/lobby/e/ng " )
self . assert_path ( resp , " response/lobby/e/uid " )
self . assert_path ( resp , " response/lobby/e/pn " )
self . assert_path ( resp , " response/lobby/e/uattr " )
self . assert_path ( resp , " response/lobby/e/mopt " )
self . assert_path ( resp , " response/lobby/e/mg " )
self . assert_path ( resp , " response/lobby/e/tid " )
self . assert_path ( resp , " response/lobby/e/tn " )
self . assert_path ( resp , " response/lobby/e/topt " )
self . assert_path ( resp , " response/lobby/e/lid " )
self . assert_path ( resp , " response/lobby/e/sn " )
self . assert_path ( resp , " response/lobby/e/pref " )
self . assert_path ( resp , " response/lobby/e/stg " )
self . assert_path ( resp , " response/lobby/e/pside " )
self . assert_path ( resp , " response/lobby/e/eatime " )
self . assert_path ( resp , " response/lobby/e/ga " )
self . assert_path ( resp , " response/lobby/e/gp " )
self . assert_path ( resp , " response/lobby/e/la " )
return resp . child_value ( ' lobby/eid ' )
def verify_lobby_delete ( self , eid : int ) - > None :
call = self . call_node ( )
lobby = Node . void ( ' lobby ' )
lobby . set_attribute ( ' method ' , ' delete ' )
lobby . add_child ( Node . s32 ( ' eid ' , eid ) )
call . add_child ( lobby )
# Swap with server
resp = self . exchange ( ' ' , call )
# Verify that response is correct
self . assert_path ( resp , " response/lobby " )
def verify_event_w_update_status ( self , loc : str , extid : int ) - > None :
call = self . call_node ( )
event_w = Node . void ( ' event_w ' )
call . add_child ( event_w )
event_w . set_attribute ( ' method ' , ' update_status ' )
event_w . add_child ( Node . s32 ( ' uid ' , extid ) )
event_w . add_child ( Node . string ( ' p_name ' , self . NAME ) )
event_w . add_child ( Node . s32 ( ' exp ' , 0 ) )
event_w . add_child ( Node . s32 ( ' customize ' , 0 ) )
event_w . add_child ( Node . s32 ( ' tid ' , 0 ) )
event_w . add_child ( Node . string ( ' t_name ' , ' ' ) )
event_w . add_child ( Node . string ( ' lid ' , loc ) )
event_w . add_child ( Node . string ( ' s_name ' , ' ' ) )
event_w . add_child ( Node . s8 ( ' pref ' , 51 ) )
event_w . add_child ( Node . s32 ( ' time ' , Time . now ( ) ) )
event_w . add_child ( Node . s8 ( ' status ' , 1 ) )
event_w . add_child ( Node . s8 ( ' stage ' , 0 ) )
event_w . add_child ( Node . s32 ( ' mid ' , - 1 ) )
event_w . add_child ( Node . s8 ( ' ng ' , - 1 ) )
# Swap with server
resp = self . exchange ( ' ' , call )
# Verify that response is correct
self . assert_path ( resp , " response/event_w/@status " )
def verify_event_w_add_comment ( self , loc : str , extid : int ) - > None :
call = self . call_node ( )
event_w = Node . void ( ' event_w ' )
call . add_child ( event_w )
event_w . set_attribute ( ' method ' , ' add_comment ' )
event_w . add_child ( Node . s32 ( ' uid ' , extid ) )
event_w . add_child ( Node . string ( ' p_name ' , self . NAME ) )
event_w . add_child ( Node . s32 ( ' exp ' , 0 ) )
event_w . add_child ( Node . s32 ( ' customize ' , 0 ) )
event_w . add_child ( Node . s32 ( ' tid ' , 0 ) )
event_w . add_child ( Node . string ( ' t_name ' , ' ' ) )
event_w . add_child ( Node . string ( ' lid ' , loc ) )
event_w . add_child ( Node . string ( ' s_name ' , ' ' ) )
event_w . add_child ( Node . s8 ( ' pref ' , 51 ) )
event_w . add_child ( Node . s32 ( ' time ' , Time . now ( ) ) )
event_w . add_child ( Node . string ( ' comment ' , ' アメ〜〜! ' ) )
event_w . add_child ( Node . bool ( ' is_tweet ' , False ) )
# Swap with server
resp = self . exchange ( ' ' , call )
# Verify that response is correct
self . assert_path ( resp , " response/event_w/@status " )
def verify_event_r_get_all ( self , extid : int ) - > None :
call = self . call_node ( )
event_r = Node . void ( ' event_r ' )
call . add_child ( event_r )
event_r . set_attribute ( ' method ' , ' get_all ' )
event_r . add_child ( Node . s32 ( ' uid ' , extid ) )
event_r . add_child ( Node . s32 ( ' time ' , 0 ) )
event_r . add_child ( Node . s32 ( ' limit ' , 30 ) )
# Swap with server
resp = self . exchange ( ' ' , call )
# Verify that response is correct. We should have at least one
# comment (the one we just wrote) and one status (because we just
# called update_status).
self . assert_path ( resp , " response/event_r/time " )
self . assert_path ( resp , " response/event_r/status/s/uid " )
self . assert_path ( resp , " response/event_r/status/s/p_name " )
self . assert_path ( resp , " response/event_r/status/s/exp " )
self . assert_path ( resp , " response/event_r/status/s/customize " )
self . assert_path ( resp , " response/event_r/status/s/tid " )
self . assert_path ( resp , " response/event_r/status/s/t_name " )
self . assert_path ( resp , " response/event_r/status/s/lid " )
self . assert_path ( resp , " response/event_r/status/s/s_name " )
self . assert_path ( resp , " response/event_r/status/s/pref " )
self . assert_path ( resp , " response/event_r/status/s/time " )
self . assert_path ( resp , " response/event_r/status/s/status " )
self . assert_path ( resp , " response/event_r/status/s/stage " )
self . assert_path ( resp , " response/event_r/status/s/mid " )
self . assert_path ( resp , " response/event_r/status/s/ng " )
self . assert_path ( resp , " response/event_r/comment/c/uid " )
self . assert_path ( resp , " response/event_r/comment/c/p_name " )
self . assert_path ( resp , " response/event_r/comment/c/exp " )
self . assert_path ( resp , " response/event_r/comment/c/customize " )
self . assert_path ( resp , " response/event_r/comment/c/tid " )
self . assert_path ( resp , " response/event_r/comment/c/t_name " )
self . assert_path ( resp , " response/event_r/comment/c/lid " )
self . assert_path ( resp , " response/event_r/comment/c/s_name " )
self . assert_path ( resp , " response/event_r/comment/c/pref " )
self . assert_path ( resp , " response/event_r/comment/c/time " )
self . assert_path ( resp , " response/event_r/comment/c/comment " )
self . assert_path ( resp , " response/event_r/comment/c/is_tweet " )
# Verify we posted our comment earlier
found = False
for child in resp . child ( ' event_r/comment ' ) . children :
if child . name != ' c ' :
continue
if child . child_value ( ' uid ' ) == extid :
name = child . child_value ( ' p_name ' )
comment = child . child_value ( ' comment ' )
if name != self . NAME :
2020-01-07 22:29:07 +01:00
raise Exception ( f ' Invalid name \' { name } \' returned for comment! ' )
2019-12-08 22:43:49 +01:00
if comment != ' アメ〜〜! ' :
2020-01-07 22:29:07 +01:00
raise Exception ( f ' Invalid comment \' { comment } \' returned for comment! ' )
2019-12-08 22:43:49 +01:00
found = True
if not found :
raise Exception ( ' Comment we posted was not found! ' )
# Verify our status came through
found = False
for child in resp . child ( ' event_r/status ' ) . children :
if child . name != ' s ' :
continue
if child . child_value ( ' uid ' ) == extid :
name = child . child_value ( ' p_name ' )
if name != self . NAME :
2020-01-07 22:29:07 +01:00
raise Exception ( f ' Invalid name \' { name } \' returned for status! ' )
2019-12-08 22:43:49 +01:00
found = True
if not found :
raise Exception ( ' Status was not found! ' )
def verify ( self , cardid : Optional [ str ] ) - > None :
# Verify boot sequence is okay
self . verify_services_get (
expected_services = [
' pcbtracker ' ,
' pcbevent ' ,
' local ' ,
' message ' ,
' facility ' ,
' cardmng ' ,
' package ' ,
' posevent ' ,
' pkglist ' ,
' dlstatus ' ,
' eacoin ' ,
' lobby ' ,
' ntp ' ,
' keepalive '
]
)
paseli_enabled = self . verify_pcbtracker_alive ( )
self . verify_message_get ( )
self . verify_package_list ( )
location = self . verify_facility_get ( )
self . verify_pcbevent_put ( )
self . verify_log_pcb_status ( location )
self . verify_pcbinfo_get ( location )
self . verify_sysinfo_get ( )
self . verify_ranking_read ( )
# Verify card registration and profile lookup
if cardid is not None :
card = cardid
else :
card = self . random_card ( )
2020-01-07 22:29:07 +01:00
print ( f " Generated random card ID { card } for use. " )
2019-12-08 22:43:49 +01:00
if cardid is None :
self . verify_cardmng_inquire ( card , msg_type = ' unregistered ' , paseli_enabled = paseli_enabled )
ref_id = self . verify_cardmng_getrefid ( card )
if len ( ref_id ) != 16 :
2020-01-07 22:29:07 +01:00
raise Exception ( f ' Invalid refid \' { ref_id } \' returned when registering card ' )
2019-12-08 22:43:49 +01:00
if ref_id != self . verify_cardmng_inquire ( card , msg_type = ' new ' , paseli_enabled = paseli_enabled ) :
2020-01-07 22:29:07 +01:00
raise Exception ( f ' Invalid refid \' { ref_id } \' returned when querying card ' )
2019-12-08 22:43:49 +01:00
# Always get a player start, regardless of new profile or not
self . verify_player_start ( ref_id )
self . verify_player_delete ( ref_id )
extid = self . verify_player_write (
ref_id ,
0 ,
location ,
[ ] ,
[ ] ,
)
else :
print ( " Skipping new card checks for existing card " )
ref_id = self . verify_cardmng_inquire ( card , msg_type = ' query ' , paseli_enabled = paseli_enabled )
# Verify pin handling and return card handling
self . verify_cardmng_authpass ( ref_id , correct = True )
self . verify_cardmng_authpass ( ref_id , correct = False )
if ref_id != self . verify_cardmng_inquire ( card , msg_type = ' query ' , paseli_enabled = paseli_enabled ) :
2020-01-07 22:29:07 +01:00
raise Exception ( f ' Invalid refid \' { ref_id } \' returned when querying card ' )
2019-12-08 22:43:49 +01:00
# Verify lobby functionality
self . verify_lobby_read ( location , extid )
eid = self . verify_lobby_entry ( location , extid )
self . verify_lobby_delete ( eid )
# Verify status updates and puzzle comments
self . verify_event_w_update_status ( location , extid )
self . verify_event_w_add_comment ( location , extid )
self . verify_event_r_get_all ( extid )
# Limelight is weird and sends only the top record for each song you played,
# and then a separate battle log. So, emulating that is kinda hard.
scores : List [ Dict [ str , int ] ] = [ ]
if cardid is None :
# Verify score saving and updating
for phase in [ 1 , 2 ] :
if phase == 1 :
dummyscores = [
# An okay score on a chart
{
' id ' : 1 ,
' chart ' : 1 ,
' clear_type ' : 2 ,
' achievement_rate ' : 7543 ,
' score ' : 432 ,
' combo ' : 123 ,
' miss_count ' : 5 ,
} ,
# A good score on an easier chart of the same song
{
' id ' : 1 ,
' chart ' : 0 ,
' clear_type ' : 3 ,
' achievement_rate ' : 9876 ,
' score ' : 543 ,
' combo ' : 543 ,
' miss_count ' : 0 ,
} ,
# A bad score on a hard chart
{
' id ' : 3 ,
' chart ' : 2 ,
' clear_type ' : 2 ,
' achievement_rate ' : 1234 ,
' score ' : 123 ,
' combo ' : 42 ,
' miss_count ' : 54 ,
} ,
# A terrible score on an easy chart
{
' id ' : 3 ,
' chart ' : 0 ,
' clear_type ' : 2 ,
' achievement_rate ' : 1024 ,
' score ' : 50 ,
' combo ' : 12 ,
' miss_count ' : 90 ,
} ,
]
if phase == 2 :
dummyscores = [
# A better score on the same chart
{
' id ' : 1 ,
' chart ' : 1 ,
' clear_type ' : 3 ,
' achievement_rate ' : 8765 ,
' score ' : 469 ,
' combo ' : 468 ,
' miss_count ' : 1 ,
} ,
# A worse score on another same chart
{
' id ' : 1 ,
' chart ' : 0 ,
' clear_type ' : 2 ,
' achievement_rate ' : 8765 ,
' score ' : 432 ,
' combo ' : 321 ,
' miss_count ' : 15 ,
' expected_score ' : 543 ,
' expected_clear_type ' : 3 ,
' expected_achievement_rate ' : 9876 ,
' expected_combo ' : 543 ,
' expected_miss_count ' : 0 ,
} ,
]
self . verify_player_write ( ref_id , extid , location , scores , dummyscores )
self . verify_log_play ( extid , location , dummyscores )
scores = self . verify_player_read ( ref_id , location )
for expected in dummyscores :
actual = None
for received in scores :
if received [ ' id ' ] == expected [ ' id ' ] and received [ ' chart ' ] == expected [ ' chart ' ] :
actual = received
break
if actual is None :
2020-01-07 22:29:07 +01:00
raise Exception ( f " Didn ' t find song { expected [ ' id ' ] } chart { expected [ ' chart ' ] } in response! " )
2019-12-08 22:43:49 +01:00
if ' expected_score ' in expected :
expected_score = expected [ ' expected_score ' ]
else :
expected_score = expected [ ' score ' ]
if ' expected_achievement_rate ' in expected :
expected_achievement_rate = expected [ ' expected_achievement_rate ' ]
else :
expected_achievement_rate = expected [ ' achievement_rate ' ]
if ' expected_clear_type ' in expected :
expected_clear_type = expected [ ' expected_clear_type ' ]
else :
expected_clear_type = expected [ ' clear_type ' ]
if ' expected_combo ' in expected :
expected_combo = expected [ ' expected_combo ' ]
else :
expected_combo = expected [ ' combo ' ]
if ' expected_miss_count ' in expected :
expected_miss_count = expected [ ' expected_miss_count ' ]
else :
expected_miss_count = expected [ ' miss_count ' ]
if actual [ ' score ' ] != expected_score :
2020-01-07 22:29:07 +01:00
raise Exception ( f ' Expected a score of \' { expected_score } \' for song \' { expected [ " id " ] } \' chart \' { expected [ " chart " ] } \' but got score \' { actual [ " score " ] } \' ' )
2019-12-08 22:43:49 +01:00
if actual [ ' achievement_rate ' ] != expected_achievement_rate :
2020-01-07 22:29:07 +01:00
raise Exception ( f ' Expected an achievement rate of \' { expected_achievement_rate } \' for song \' { expected [ " id " ] } \' chart \' { expected [ " chart " ] } \' but got achievement rate \' { actual [ " achievement_rate " ] } \' ' )
2019-12-08 22:43:49 +01:00
if actual [ ' clear_type ' ] != expected_clear_type :
2020-01-07 22:29:07 +01:00
raise Exception ( f ' Expected a clear_type of \' { expected_clear_type } \' for song \' { expected [ " id " ] } \' chart \' { expected [ " chart " ] } \' but got clear_type \' { actual [ " clear_type " ] } \' ' )
2019-12-08 22:43:49 +01:00
if actual [ ' combo ' ] != expected_combo :
2020-01-07 22:29:07 +01:00
raise Exception ( f ' Expected a combo of \' { expected_combo } \' for song \' { expected [ " id " ] } \' chart \' { expected [ " chart " ] } \' but got combo \' { actual [ " combo " ] } \' ' )
2019-12-08 22:43:49 +01:00
if actual [ ' miss_count ' ] != expected_miss_count :
2020-01-07 22:29:07 +01:00
raise Exception ( f ' Expected a miss count of \' { expected_miss_count } \' for song \' { expected [ " id " ] } \' chart \' { expected [ " chart " ] } \' but got miss count \' { actual [ " miss_count " ] } \' ' )
2019-12-08 22:43:49 +01:00
# Sleep so we don't end up putting in score history on the same second
time . sleep ( 1 )
else :
print ( " Skipping score checks for existing card " )
# Verify ending game
self . verify_player_end ( ref_id )
# Verify paseli handling
if paseli_enabled :
print ( " PASELI enabled for this PCBID, executing PASELI checks " )
else :
print ( " PASELI disabled for this PCBID, skipping PASELI checks " )
return
sessid , balance = self . verify_eacoin_checkin ( card )
if balance == 0 :
print ( " Skipping PASELI consume check because card has 0 balance " )
else :
self . verify_eacoin_consume ( sessid , balance , random . randint ( 0 , balance ) )
self . verify_eacoin_checkout ( sessid )