Changed how environment variables are set in the Docker container.master
parent
92ccfdccf9
commit
02d1aeb408
@ -1,5 +1,5 @@ |
||||
[Debug] |
||||
|
||||
; NOTE: These need to be mapped in if you want to run the test suite against a container. |
||||
TEST_VASSAL_ENGINES = /test-data/vassal/ |
||||
TEST_VASL_MODS = /test-data/vasl-vmods/ |
||||
TEST_VASL_EXTNS_DIR = /test-data/vasl-extensions/ |
||||
TEST_VASL_MODS = /test-data/vasl-mods/ |
||||
|
||||
@ -0,0 +1,10 @@ |
||||
[Debug] |
||||
|
||||
; Set this if you want to run the test suite (allows the webapp server to be controlled using gRPC). |
||||
; CONTROL_TESTS_PORT = -1 |
||||
|
||||
; Set this to a directory containing the VASSAL releases to run the test suite with. |
||||
; TEST_VASSAL_ENGINES = ... |
||||
|
||||
; Set this to a directory containing the VASL modules (.vmod files) to run the test suite with. |
||||
; TEST_VASL_MODS = ... |
||||
@ -1,45 +0,0 @@ |
||||
"""Webapp handlers for testing porpoises.""" |
||||
|
||||
import inspect |
||||
import base64 |
||||
|
||||
from flask import request, jsonify, abort |
||||
|
||||
from vasl_templates.webapp import app |
||||
from vasl_templates.webapp.tests.remote import ControlTests |
||||
|
||||
_control_tests = ControlTests( app ) |
||||
|
||||
# --------------------------------------------------------------------- |
||||
|
||||
@app.route( "/control-tests/<action>" ) |
||||
def control_tests( action ): |
||||
"""Accept commands from a remote test suite.""" |
||||
|
||||
# check if this functionality has been enabled |
||||
if not app.config.get( "ENABLE_REMOTE_TEST_CONTROL" ): |
||||
abort( 404 ) |
||||
|
||||
# figure out what we're being asked to do |
||||
func = getattr( _control_tests, action ) |
||||
if not func: |
||||
abort( 404 ) |
||||
|
||||
# get any parameters |
||||
sig = inspect.signature( func ) |
||||
kwargs = {} |
||||
for param in sig.parameters.values(): |
||||
if param.name in ("vengine","vmod","gpids","key","val","dtype","fname","dname","extns_dtype","bin_data"): |
||||
kwargs[ param.name ] = request.args.get( param.name, param.default ) |
||||
if param.name == "bin_data" and kwargs["bin_data"]: |
||||
kwargs["bin_data"] = base64.b64decode( kwargs["bin_data"] ) |
||||
|
||||
# execute the command |
||||
resp = func( **kwargs ) |
||||
|
||||
# return any response |
||||
if isinstance( resp, (str,list,dict) ): |
||||
return jsonify( resp ) |
||||
else: |
||||
assert resp == _control_tests, "Methods should return self if there is no response data." |
||||
return "ok" |
||||
@ -0,0 +1,17 @@ |
||||
cd `dirname "$0"`/proto |
||||
|
||||
# initialize |
||||
rm -rf generated 2>/dev/null |
||||
mkdir generated |
||||
|
||||
# compile the protobuf definitions |
||||
python -m grpc_tools.protoc \ |
||||
--proto_path . \ |
||||
--python_out=generated/ \ |
||||
--grpc_python_out=generated/ \ |
||||
control_tests.proto |
||||
|
||||
# FUDGE! Fix a bogus import :-/ |
||||
sed --in-place \ |
||||
's/^import control_tests_pb2 as control__tests__pb2$/import vasl_templates.webapp.tests.proto.generated.control_tests_pb2 as control__tests__pb2/' \ |
||||
generated/control_tests_pb2_grpc.py |
||||
@ -0,0 +1,227 @@ |
||||
""" Allow the test suite to control a remote webapp server. """ |
||||
|
||||
import json |
||||
import base64 |
||||
|
||||
import grpc |
||||
from google.protobuf.empty_pb2 import Empty |
||||
|
||||
from vasl_templates.webapp.tests.proto.generated.control_tests_pb2_grpc import ControlTestsStub |
||||
from vasl_templates.webapp.tests.proto.utils import enum_from_string |
||||
|
||||
from vasl_templates.webapp.tests.proto.generated.control_tests_pb2 import \ |
||||
SetVassalVersionRequest, SetVaslVersionRequest, SetVaslExtnInfoDirRequest, SetGpidRemappingsRequest, \ |
||||
SetDataDirRequest, SetDefaultScenarioRequest, SetDefaultTemplatePackRequest, \ |
||||
SetVehOrdNotesDirRequest, SetUserFilesDirRequest, \ |
||||
SetAsaScenarioIndexRequest, SetRoarScenarioIndexRequest, \ |
||||
DumpVsavRequest, GetVaslPiecesRequest, \ |
||||
SetAppConfigValRequest, DeleteAppConfigValRequest, \ |
||||
SaveTempFileRequest |
||||
|
||||
# --------------------------------------------------------------------- |
||||
|
||||
# NOTE: The API for this class should be kept in sync with ControlTestsServicer. |
||||
|
||||
class ControlTests: #pylint: disable=too-many-public-methods |
||||
"""Control a remote webapp server.""" |
||||
|
||||
def __init__( self, addr ): |
||||
# initialize |
||||
channel = grpc.insecure_channel( addr ) |
||||
self._stub = ControlTestsStub( channel ) |
||||
self._caps = None |
||||
|
||||
def has_capability( self, cap ) : |
||||
"""Check if the remote webapp has the specified capability.""" |
||||
return cap in self._caps |
||||
|
||||
def start_tests( self ): |
||||
"""Start a new test run.""" |
||||
resp = self._stub.startTests( Empty() ) |
||||
self._caps = set( resp.capabilities ) |
||||
return self |
||||
|
||||
def end_tests( self ): |
||||
"""End a test run.""" |
||||
self._stub.endTests( Empty() ) |
||||
self._caps = None |
||||
|
||||
def get_vassal_versions( self ): |
||||
"""Get the available VASSAL versions.""" |
||||
resp = self._stub.getVassalVersions( Empty() ) |
||||
return resp.vassalVersions |
||||
|
||||
def set_vassal_version( self, vassal_version ): |
||||
"""Set the VASSAL version.""" |
||||
self._stub.setVassalVersion( |
||||
SetVassalVersionRequest( vassalVersion = vassal_version ) |
||||
) |
||||
return self |
||||
|
||||
def get_vasl_versions( self ): |
||||
"""Get the available VASL versions.""" |
||||
resp = self._stub.getVaslVersions( Empty() ) |
||||
return resp.vaslVersions |
||||
|
||||
def set_vasl_version( self, vasl_mod, vasl_extns_type ): |
||||
"""Set the VASL version.""" |
||||
vasl_extns_type = enum_from_string( |
||||
SetVaslVersionRequest.VaslExtnsType, #pylint: disable=no-member |
||||
vasl_extns_type or "{NONE}" |
||||
) |
||||
self._stub.setVaslVersion( |
||||
SetVaslVersionRequest( vaslVersion=vasl_mod, vaslExtnsType=vasl_extns_type ) |
||||
) |
||||
return self |
||||
|
||||
def get_vasl_extns( self ): |
||||
"""Get the VASL extensions.""" |
||||
resp = self._stub.getVaslExtns( Empty() ) |
||||
return json.loads( resp.vaslExtnsJson ) |
||||
|
||||
def set_vasl_extn_info_dir( self, dname ): |
||||
"""Set the VASL extensions info directory.""" |
||||
self._stub.setVaslExtnInfoDir( |
||||
SetVaslExtnInfoDirRequest( dirName = dname ) |
||||
) |
||||
return self |
||||
|
||||
def set_gpid_remappings( self, gpid_remappings ): |
||||
"""Set the GPID remappings.""" |
||||
self._stub.setGpidRemappings( |
||||
SetGpidRemappingsRequest( gpidRemappingsJson = json.dumps( gpid_remappings ) ) |
||||
) |
||||
return self |
||||
|
||||
def get_vasl_mod_warnings( self ): |
||||
"""Get the vasl_mod warnings.""" |
||||
resp = self._stub.getVaslModWarnings( Empty() ) |
||||
return resp.warnings |
||||
|
||||
def set_data_dir( self, dtype ): |
||||
"""Set the data directory.""" |
||||
dtype = enum_from_string( SetDataDirRequest.DirType, dtype ) #pylint: disable=no-member |
||||
self._stub.setDataDir( |
||||
SetDataDirRequest( dirType = dtype ) |
||||
) |
||||
return self |
||||
|
||||
def set_default_scenario( self, fname ): |
||||
"""Set the default scenario.""" |
||||
self._stub.setDefaultScenario( |
||||
SetDefaultScenarioRequest( fileName = fname ) |
||||
) |
||||
return self |
||||
|
||||
def set_default_template_pack( self, template_pack ): |
||||
"""Set the default template pack.""" |
||||
if isinstance( template_pack, str ) and template_pack.startswith( "{" ) and template_pack.endswith( "}" ): |
||||
val = enum_from_string( |
||||
SetDefaultTemplatePackRequest.TemplatePackType, #pylint: disable=no-member |
||||
template_pack |
||||
) |
||||
req = SetDefaultTemplatePackRequest( templatePackType = val ) |
||||
elif isinstance( template_pack, str ): |
||||
req = SetDefaultTemplatePackRequest( dirName = template_pack ) |
||||
elif isinstance( template_pack, bytes ): |
||||
req = SetDefaultTemplatePackRequest( zipData = template_pack ) |
||||
else: |
||||
raise ValueError( "Can't identify template pack type: {}".format( type(template_pack).__name__ ) ) |
||||
self._stub.setDefaultTemplatePack( req ) |
||||
return self |
||||
|
||||
def set_vo_notes_dir( self, dtype ): |
||||
"""Set the vehicle/ordnance notes directory.""" |
||||
dtype = enum_from_string( SetVehOrdNotesDirRequest.DirType, dtype or "{NONE}" ) #pylint: disable=no-member |
||||
self._stub.setVehOrdNotesDir( |
||||
SetVehOrdNotesDirRequest( dirType = dtype ) |
||||
) |
||||
return self |
||||
|
||||
def set_user_files_dir( self, dname_or_url ): |
||||
"""Set the user files directory.""" |
||||
self._stub.setUserFilesDir( |
||||
SetUserFilesDirRequest( dirOrUrl = dname_or_url ) |
||||
) |
||||
return self |
||||
|
||||
def set_asa_scenario_index( self, fname ): |
||||
"""Set the ASL Scenario Archive scenario index.""" |
||||
self._stub.setAsaScenarioIndex( |
||||
SetAsaScenarioIndexRequest( fileName = fname ) |
||||
) |
||||
return self |
||||
|
||||
def set_roar_scenario_index( self, fname ): |
||||
"""Set the ROAR scenario index.""" |
||||
self._stub.setRoarScenarioIndex( |
||||
SetRoarScenarioIndexRequest( fileName = fname ) |
||||
) |
||||
return self |
||||
|
||||
def get_last_snippet_image( self ): |
||||
"""Get the last snippet image.""" |
||||
resp = self._stub.getLastSnippetImage( Empty() ) |
||||
return resp.imageData |
||||
|
||||
def reset_last_asa_upload( self ): |
||||
"""Reset the last ASL Scenario Archive upload.""" |
||||
self._stub.resetLastAsaUpload( Empty() ) |
||||
return self |
||||
|
||||
def get_last_asa_upload( self ): |
||||
"""Get the last ASL Scenario Archive upload.""" |
||||
resp = self._stub.getLastAsaUpload( Empty() ) |
||||
last_asa_upload = json.loads( resp.lastUploadJson ) |
||||
if last_asa_upload: |
||||
for key in ("vasl_setup","screenshot"): |
||||
if last_asa_upload.get( key ): |
||||
last_asa_upload[key] = base64.b64decode( last_asa_upload[key].encode( "ascii" ) ) |
||||
return last_asa_upload |
||||
|
||||
def dump_vsav( self, vsav ): |
||||
"""Dump a VASL save file.""" |
||||
if isinstance( vsav, str ): |
||||
with open( vsav, "rb" ) as fp: |
||||
vsav = fp.read() |
||||
resp = self._stub.dumpVsav( |
||||
DumpVsavRequest( vsavData = vsav ) |
||||
) |
||||
return resp.vsavDump |
||||
|
||||
def get_vasl_pieces( self, vasl_version ): |
||||
"""Get the pieces for the specified VASL module.""" |
||||
resp = self._stub.getVaslPieces( GetVaslPiecesRequest( vaslVersion=vasl_version ) ) |
||||
return resp.pieceDump, resp.gpids |
||||
|
||||
def get_app_config( self ): |
||||
"""Get the app config.""" |
||||
resp = self._stub.getAppConfig( Empty() ) |
||||
return json.loads( resp.appConfigJson ) |
||||
|
||||
def set_app_config_val( self, key, val ): |
||||
"""Set an app config value.""" |
||||
if isinstance( val, str ): |
||||
req = SetAppConfigValRequest( key=key, strVal=val ) |
||||
elif isinstance( val, int ): |
||||
req = SetAppConfigValRequest( key=key, intVal=val ) |
||||
elif isinstance( val, bool ): |
||||
req = SetAppConfigValRequest( key=key, boolVal=val ) |
||||
else: |
||||
raise ValueError( "Invalid value type: {}".format( type(val).__name__ ) ) |
||||
self._stub.setAppConfigVal( req ) |
||||
return self |
||||
|
||||
def delete_app_config_key( self, key ): |
||||
"""Delete an app config value.""" |
||||
self._stub.deleteAppConfigVal( |
||||
DeleteAppConfigValRequest( key = key ) |
||||
) |
||||
return self |
||||
|
||||
def save_temp_file( self, fname, data ): |
||||
"""Save a temp file.""" |
||||
self._stub.saveTempFile( |
||||
SaveTempFileRequest( fileName=fname, data=data ) |
||||
) |
||||
return self |
||||
@ -0,0 +1,536 @@ |
||||
"""gRPC servicer that allows the webapp server to be controlled.""" |
||||
|
||||
import os |
||||
import json |
||||
import tempfile |
||||
import glob |
||||
import re |
||||
import logging |
||||
import io |
||||
import copy |
||||
import base64 |
||||
import inspect |
||||
import random |
||||
|
||||
import tabulate |
||||
from google.protobuf.empty_pb2 import Empty |
||||
|
||||
from vasl_templates.webapp.config.constants import DATA_DIR |
||||
from vasl_templates.webapp.vassal import VassalShim |
||||
from vasl_templates.webapp.utils import TempFile |
||||
from vasl_templates.webapp import \ |
||||
main as webapp_main, \ |
||||
vasl_mod as webapp_vasl_mod, \ |
||||
scenarios as webapp_scenarios, \ |
||||
snippets as webapp_snippets, \ |
||||
globvars as webapp_globvars |
||||
|
||||
from vasl_templates.webapp.tests.proto.generated.control_tests_pb2_grpc \ |
||||
import ControlTestsServicer as BaseControlTestsServicer |
||||
|
||||
from vasl_templates.webapp.tests.proto.generated.control_tests_pb2 import \ |
||||
SetVassalVersionRequest, SetVaslVersionRequest, SetVaslExtnInfoDirRequest, SetGpidRemappingsRequest, \ |
||||
SetDataDirRequest, SetDefaultScenarioRequest, SetDefaultTemplatePackRequest, \ |
||||
SetVehOrdNotesDirRequest, SetUserFilesDirRequest, \ |
||||
SetAsaScenarioIndexRequest, SetRoarScenarioIndexRequest, \ |
||||
SetAppConfigValRequest, DeleteAppConfigValRequest |
||||
from vasl_templates.webapp.tests.proto.generated.control_tests_pb2 import \ |
||||
StartTestsResponse, \ |
||||
GetVassalVersionsResponse, GetVaslVersionsResponse, GetVaslExtnsResponse, GetVaslModWarningsResponse, \ |
||||
GetLastSnippetImageResponse, GetLastAsaUploadResponse, \ |
||||
DumpVsavResponse, GetVaslPiecesResponse, GetAppConfigResponse |
||||
|
||||
# nb: these are defined as a convenience |
||||
_VaslExtnsTypes_NONE = SetVaslVersionRequest.VaslExtnsType.NONE #pylint: disable=no-member |
||||
_VaslExtnsTypes_REAL = SetVaslVersionRequest.VaslExtnsType.REAL #pylint: disable=no-member |
||||
_VaslExtnsTypes_TEMP_DIR = SetVaslVersionRequest.VaslExtnsType.TEMP_DIR #pylint: disable=no-member |
||||
_TemplatePackTypes_DEFAULT = SetDefaultTemplatePackRequest.TemplatePackType.DEFAULT #pylint: disable=no-member |
||||
_TemplatePackTypes_REAL = SetDefaultTemplatePackRequest.TemplatePackType.REAL #pylint: disable=no-member |
||||
|
||||
_logger = logging.getLogger( "control_tests" ) |
||||
|
||||
_FIXTURES_DIR = os.path.join( os.path.dirname(__file__), "fixtures" ) |
||||
_ORIG_GPID_REMAPPINGS = copy.deepcopy( webapp_vasl_mod.GPID_REMAPPINGS ) |
||||
_ORIG_CHAPTER_H_NOTES_DIR = None |
||||
|
||||
# --------------------------------------------------------------------- |
||||
|
||||
# NOTE: The API for this class should be kept in sync with ControlTests. |
||||
|
||||
class ControlTestsServicer( BaseControlTestsServicer ): #pylint: disable=too-many-public-methods |
||||
"""Allows a webapp server to be controlled by a remote client.""" |
||||
|
||||
def __init__( self, webapp ): |
||||
|
||||
# initialize |
||||
self._webapp = webapp |
||||
global _ORIG_CHAPTER_H_NOTES_DIR |
||||
if not _ORIG_CHAPTER_H_NOTES_DIR: |
||||
_ORIG_CHAPTER_H_NOTES_DIR = webapp.config.get( "CHAPTER_H_NOTES_DIR" ) |
||||
self._temp_dir = None |
||||
|
||||
# look for VASSAL engines |
||||
_logger.debug( "Locating VASSAL engines:" ) |
||||
self._vassal_engines = {} |
||||
dname = self._webapp.config.get( "TEST_VASSAL_ENGINES" ) |
||||
if dname: |
||||
for root,_,fnames in os.walk( dname ): |
||||
if os.sep + "_disabled_" + os.sep in root: |
||||
continue |
||||
for fname in fnames: |
||||
if fname == "Vengine.jar": |
||||
if root.endswith( "/lib" ): |
||||
root = root[:-4] |
||||
# FUDGE! We assume that the version number is part of the path (we can do this |
||||
# since we are only used for running tests i.e. in a controlled environment). |
||||
mo = re.search( r"\d+\.\d+\.\d+", root ) |
||||
self._vassal_engines[ mo.group() ] = root |
||||
break |
||||
for key,val in self._vassal_engines.items(): |
||||
_logger.debug( "- %s -> %s", key, val ) |
||||
|
||||
# look for VASL modules |
||||
_logger.debug( "Locating VASL modules:" ) |
||||
self._vasl_mods = {} |
||||
dname = self._webapp.config.get( "TEST_VASL_MODS" ) |
||||
if dname: |
||||
fspec = os.path.join( dname, "*.vmod" ) |
||||
for fname in glob.glob( fspec ): |
||||
# FUDGE! We assume that the version number is part of the filename (we can do this |
||||
# since we are only used for running tests i.e. in a controlled environment). |
||||
mo = re.search( r"\d+\.\d+\.\d+", os.path.basename(fname) ) |
||||
self._vasl_mods[ mo.group() ] = fname |
||||
for key,val in self._vasl_mods.items(): |
||||
_logger.debug( "- %s -> %s", key, val ) |
||||
|
||||
def __del__( self ): |
||||
# clean up |
||||
self.cleanup() |
||||
|
||||
def cleanup( self ): |
||||
"""Clean up.""" |
||||
if self._temp_dir: |
||||
self._temp_dir.cleanup() |
||||
self._temp_dir = None |
||||
|
||||
def startTests( self, request, context ): |
||||
"""Start a new test run.""" |
||||
_logger.info( "=== START TESTS ===" ) |
||||
|
||||
# check that everything has been configured properly |
||||
# NOTE: We do this here instead of __init__() so that we can return an error message to the client, |
||||
# rather than having the servicer fail to start up, giving the client a "can't connect" error. |
||||
if not self._vassal_engines: |
||||
raise RuntimeError( "No VASSAL releases were configured (see debug.cfg.example)." ) |
||||
if not self._vasl_mods: |
||||
raise RuntimeError( "No VASL modules were configured (see debug.cfg.example)." ) |
||||
|
||||
# set up a directory for our temp files |
||||
if self._temp_dir: |
||||
self._temp_dir.cleanup() |
||||
self._temp_dir = tempfile.TemporaryDirectory() |
||||
|
||||
# reset the webapp server |
||||
ctx = None |
||||
self.setDataDir( |
||||
SetDataDirRequest( dirType = SetDataDirRequest.DirType.TEST ), ctx #pylint: disable=no-member |
||||
) |
||||
self.setDefaultScenario( SetDefaultScenarioRequest( fileName=None ), ctx ) |
||||
self.setDefaultTemplatePack( |
||||
SetDefaultTemplatePackRequest( templatePackType = _TemplatePackTypes_DEFAULT ), |
||||
ctx |
||||
) |
||||
self.setVehOrdNotesDir( |
||||
SetVehOrdNotesDirRequest( dirType = SetVehOrdNotesDirRequest.DirType.NONE ), ctx #pylint: disable=no-member |
||||
) |
||||
self.setUserFilesDir( SetUserFilesDirRequest( dirOrUrl=None ), ctx ) |
||||
self.setVassalVersion( SetVassalVersionRequest( vassalVersion=None ), ctx ) |
||||
self.setVaslVersion( SetVaslVersionRequest( vaslVersion=None ), ctx ) |
||||
self.setGpidRemappings( |
||||
SetGpidRemappingsRequest( gpidRemappingsJson = json.dumps(_ORIG_GPID_REMAPPINGS) ), ctx |
||||
) |
||||
self.setVaslExtnInfoDir( SetVaslExtnInfoDirRequest( dirName=None ), ctx ) |
||||
self.setAsaScenarioIndex( SetAsaScenarioIndexRequest( fileName="asl-scenario-archive.json" ), ctx ) |
||||
self.setRoarScenarioIndex( SetRoarScenarioIndexRequest( fileName="roar-scenario-index.json" ), ctx ) |
||||
self.setAppConfigVal( SetAppConfigValRequest( key="MAP_URL", strVal="MAP:[{LAT},{LONG}]" ), ctx ) |
||||
self.setAppConfigVal( SetAppConfigValRequest( key="DISABLE_DOWNLOADED_FILES", boolVal=True ), ctx ) |
||||
self.setAppConfigVal( SetAppConfigValRequest( key="DISABLE_LOCAL_ASA_INDEX_UPDATES", boolVal=True ), ctx ) |
||||
self.setAppConfigVal( SetAppConfigValRequest( key="DISABLE_LFA_HOTNESS_FADEIN", boolVal=True ), ctx ) |
||||
self.deleteAppConfigVal( DeleteAppConfigValRequest( key="ALTERNATE_WEBAPP_BASE_URL" ), ctx ) |
||||
# NOTE: The webapp has been reconfigured, but the client must reloaed the home page |
||||
# with "?force-reinit=1", to force it to re-initialize with the new settings. |
||||
|
||||
# return our capabilities to the caller |
||||
caps = [] |
||||
if _ORIG_CHAPTER_H_NOTES_DIR: |
||||
# NOTE: Some tests require real Chapter H vehicle/ordnance notes. This is copyrighted material, |
||||
# so it is kept in a private repo. For the purpose of running tests, it is considered optional |
||||
# and tests that need it can check this capability and not run if it's not available. |
||||
caps.append( "chapter-h" ) |
||||
|
||||
return StartTestsResponse( capabilities=caps ) |
||||
|
||||
def endTests( self, request, context ): |
||||
"""End a test run.""" |
||||
self._log_request( request, context ) |
||||
# end the test run |
||||
self.cleanup() |
||||
return Empty() |
||||
|
||||
def getVassalVersions( self, request, context ): |
||||
"""Get the available VASSAL versions.""" |
||||
self._log_request( request, context ) |
||||
# get the available VASSAL versions |
||||
vassal_versions = list( self._vassal_engines.keys() ) |
||||
_logger.debug( "- Returning VASSAL versions: %s", " ; ".join( vassal_versions ) ) |
||||
return GetVassalVersionsResponse( vassalVersions=vassal_versions ) |
||||
|
||||
def setVassalVersion( self, request, context ): |
||||
"""Set the VASSAL version.""" |
||||
self._log_request( request, context ) |
||||
vassal_version = request.vassalVersion |
||||
# set the VASSAL engine |
||||
if vassal_version: |
||||
dname = self._vassal_engines.get( vassal_version ) |
||||
if not dname: |
||||
raise RuntimeError( "Unknown VASSAL version: {}".format( vassal_version ) ) |
||||
else: |
||||
dname = None |
||||
_logger.debug( "- Setting VASSAL engine: %s", dname ) |
||||
self._webapp.config[ "VASSAL_DIR" ] = dname |
||||
return Empty() |
||||
|
||||
def getVaslVersions( self, request, context ): |
||||
"""Get the available VASL versions.""" |
||||
self._log_request( request, context ) |
||||
# get the available VASL versions |
||||
vasl_versions = list( self._vasl_mods.keys() ) |
||||
_logger.debug( "- Returning VASL versions: %s", " ; ".join( vasl_versions ) ) |
||||
return GetVaslVersionsResponse( vaslVersions=vasl_versions ) |
||||
|
||||
def setVaslVersion( self, request, context ): |
||||
"""Set the VASL version.""" |
||||
self._log_request( request, context ) |
||||
vasl_version, vasl_extns_type = request.vaslVersion, request.vaslExtnsType |
||||
# set the VASL module |
||||
if vasl_version == "random": |
||||
# NOTE: Some tests require a VASL module to be loaded, and since they should all |
||||
# should behave in the same way, it doesn't matter which one we use. |
||||
fname = random.choice( list( self._vasl_mods.values() ) ) |
||||
elif vasl_version: |
||||
fname = self._vasl_mods.get( vasl_version ) |
||||
if not fname: |
||||
raise RuntimeError( "Unknown VASL version: {}".format( vasl_version ) ) |
||||
else: |
||||
fname = None |
||||
_logger.debug( "- Setting VASL module: %s", fname ) |
||||
self._webapp.config[ "VASL_MOD" ] = fname |
||||
|
||||
# configure the VASL extensions |
||||
if vasl_extns_type == _VaslExtnsTypes_NONE: |
||||
dname = None |
||||
elif vasl_extns_type == _VaslExtnsTypes_REAL: |
||||
dname = os.path.join( _FIXTURES_DIR, "vasl-extensions/real/" ) |
||||
elif vasl_extns_type == _VaslExtnsTypes_TEMP_DIR: |
||||
dname = self._temp_dir.name |
||||
else: |
||||
raise RuntimeError( "Unknown VASL extensions type: {}".format( vasl_extns_type ) ) |
||||
_logger.debug( "- Setting VASL extensions: %s", dname ) |
||||
self._webapp.config[ "VASL_EXTNS_DIR" ] = dname |
||||
|
||||
return Empty() |
||||
|
||||
def getVaslExtns( self, request, context ): |
||||
"""Get the VASL extensions.""" |
||||
self._log_request( request, context ) |
||||
# get the VASL extensions |
||||
vasl_extns = webapp_globvars.vasl_mod.get_extns() |
||||
_logger.debug( "- %s", vasl_extns ) |
||||
return GetVaslExtnsResponse( |
||||
vaslExtnsJson = json.dumps( vasl_extns ) |
||||
) |
||||
|
||||
def setVaslExtnInfoDir( self, request, context ): |
||||
"""Set the VASL extensions info directory.""" |
||||
self._log_request( request, context ) |
||||
dname = request.dirName |
||||
# set the VASL extensions info directory |
||||
if dname: |
||||
dname = os.path.join( _FIXTURES_DIR, "vasl-extensions/"+dname ) |
||||
else: |
||||
dname = None |
||||
_logger.debug( "- Setting the default VASL extension info directory: %s", dname ) |
||||
self._webapp.config[ "_VASL_EXTN_INFO_DIR_" ] = dname |
||||
return Empty() |
||||
|
||||
def setGpidRemappings( self, request, context ): |
||||
"""Set the GPID remappings.""" |
||||
self._log_request( request, context ) |
||||
gpid_remappings = json.loads( request.gpidRemappingsJson ) |
||||
# set the GPID remappings |
||||
if gpid_remappings == _ORIG_GPID_REMAPPINGS: |
||||
_logger.debug( "- Setting GPID remappings: (original)" ) |
||||
else: |
||||
_logger.debug( "- Setting GPID remappings:" ) |
||||
for vassal_version, mappings in gpid_remappings.items(): |
||||
_logger.debug( " - %s: %s", vassal_version, mappings ) |
||||
webapp_vasl_mod.GPID_REMAPPINGS = gpid_remappings |
||||
return Empty() |
||||
|
||||
def getVaslModWarnings( self, request, context ): |
||||
"""Get the vasl_mod warnings.""" |
||||
self._log_request( request, context ) |
||||
# get the vasl_mod warnings |
||||
warnings = webapp_vasl_mod._warnings #pylint: disable=protected-access |
||||
_logger.debug( "- %s", warnings ) |
||||
return GetVaslModWarningsResponse( warnings=warnings ) |
||||
|
||||
def setDataDir( self, request, context ): |
||||
"""Set the data directory.""" |
||||
self._log_request( request, context ) |
||||
dtype = request.dirType |
||||
# set the data directory |
||||
if dtype == SetDataDirRequest.DirType.TEST: #pylint: disable=no-member |
||||
dname = os.path.join( _FIXTURES_DIR, "data" ) |
||||
elif dtype == SetDataDirRequest.DirType.REAL: #pylint: disable=no-member |
||||
dname = DATA_DIR |
||||
else: |
||||
raise RuntimeError( "Unknown data dir type: {}".format( dtype ) ) |
||||
_logger.debug( "- Setting data directory: %s", dname ) |
||||
self._webapp.config[ "DATA_DIR" ] = dname |
||||
return Empty() |
||||
|
||||
def setDefaultScenario( self, request, context ): |
||||
"""Set the default scenario.""" |
||||
self._log_request( request, context ) |
||||
fname = request.fileName |
||||
# set the default scenario |
||||
if fname: |
||||
fname = os.path.join( _FIXTURES_DIR, fname ) |
||||
else: |
||||
fname = None |
||||
_logger.debug( "- Setting default scenario: %s", fname ) |
||||
webapp_main.default_scenario = fname |
||||
return Empty() |
||||
|
||||
def setDefaultTemplatePack( self, request, context ): |
||||
"""Set the default template pack.""" |
||||
self._log_request( request, context ) |
||||
# set the default template pack |
||||
if request.HasField( "templatePackType" ): |
||||
if request.templatePackType == _TemplatePackTypes_DEFAULT: |
||||
target = None |
||||
elif request.templatePackType == _TemplatePackTypes_REAL: |
||||
target = os.path.join( os.path.dirname(__file__), "../data/default-template-pack/" ) |
||||
else: |
||||
raise RuntimeError( "Invalid TemplatePackType: {}".format( request.templatePackType ) ) |
||||
elif request.HasField( "dirName" ): |
||||
target = os.path.join( _FIXTURES_DIR, "template-packs/"+request.dirName ) |
||||
elif request.HasField( "zipData" ): |
||||
fname = os.path.join( self._temp_dir.name, "default-template-pack.zip" ) |
||||
with open( fname, "wb" ) as fp: |
||||
fp.write( request.zipData ) |
||||
target = fname |
||||
else: |
||||
raise RuntimeError( "Can't find the default template pack specification." ) |
||||
_logger.debug( "- Setting default template pack: %s", target ) |
||||
webapp_snippets.default_template_pack = target |
||||
webapp_globvars.template_pack = None # nb: force the default template pack to be reloaded |
||||
return Empty() |
||||
|
||||
def setVehOrdNotesDir( self, request, context ): |
||||
"""Set the vehicle/ordnance notes directory.""" |
||||
self._log_request( request, context ) |
||||
dtype = request.dirType |
||||
# set the vehicle/ordnance notes directory |
||||
if dtype == SetVehOrdNotesDirRequest.DirType.NONE: #pylint: disable=no-member |
||||
dname = None |
||||
elif dtype == SetVehOrdNotesDirRequest.DirType.REAL: #pylint: disable=no-member |
||||
dname = _ORIG_CHAPTER_H_NOTES_DIR |
||||
elif dtype == SetVehOrdNotesDirRequest.DirType.TEST: #pylint: disable=no-member |
||||
dname = os.path.join( _FIXTURES_DIR, "vo-notes" ) |
||||
else: |
||||
raise RuntimeError( "Invalid vehicle/ordnance notes dir.type: {}".format( dtype ) ) |
||||
_logger.debug( "- Setting vehicle/ordnance notes: %s", dname ) |
||||
self._webapp.config[ "CHAPTER_H_NOTES_DIR" ] = dname |
||||
return Empty() |
||||
|
||||
def setUserFilesDir( self, request, context ): |
||||
"""Set the user files directory.""" |
||||
self._log_request( request, context ) |
||||
# set the user files directory |
||||
dname = request.dirOrUrl |
||||
if dname: |
||||
if not dname.startswith( ( "http://", "https://" ) ): |
||||
dname = os.path.join( _FIXTURES_DIR, dname ) |
||||
else: |
||||
dname = None |
||||
_logger.debug( "- Setting user files directory: %s", dname ) |
||||
self._webapp.config[ "USER_FILES_DIR" ] = dname |
||||
return Empty() |
||||
|
||||
def setAsaScenarioIndex( self, request, context ): |
||||
"""Set the ASL Scenario Archive scenario index.""" |
||||
self._log_request( request, context ) |
||||
fname = request.fileName |
||||
# set the ASL Scenario Archive scenario index |
||||
if fname: |
||||
fname = os.path.join( _FIXTURES_DIR, fname ) |
||||
else: |
||||
fname = None |
||||
_logger.debug( "- Setting ASA scenario index: %s", fname ) |
||||
webapp_scenarios._asa_scenarios._set_data( fname ) #pylint: disable=protected-access |
||||
return Empty() |
||||
|
||||
def setRoarScenarioIndex( self, request, context ): |
||||
"""Set the ROAR scenario index.""" |
||||
self._log_request( request, context ) |
||||
fname = request.fileName |
||||
# set the ROAR scenario index |
||||
if fname: |
||||
fname = os.path.join( _FIXTURES_DIR, fname ) |
||||
else: |
||||
fname = None |
||||
_logger.debug( "- Setting ROAR scenario index: %s", fname ) |
||||
webapp_scenarios._roar_scenarios._set_data( fname ) #pylint: disable=protected-access |
||||
return Empty() |
||||
|
||||
def getLastSnippetImage( self, request, context ): |
||||
"""Get the last snippet image.""" |
||||
self._log_request( request, context ) |
||||
# get the last snippet image |
||||
last_snippet_image = webapp_snippets.last_snippet_image |
||||
_logger.debug( "- Returning the last snippet image: %s", |
||||
"#bytes={}".format( len(last_snippet_image) ) if last_snippet_image else None |
||||
) |
||||
return GetLastSnippetImageResponse( imageData=last_snippet_image ) |
||||
|
||||
def resetLastAsaUpload( self, request, context ): |
||||
"""Reset the last ASL Scenario Archive upload.""" |
||||
self._log_request( request, context ) |
||||
# reset the last ASL Scenario Archive upload |
||||
webapp_scenarios._last_asa_upload = None #pylint: disable=protected-access |
||||
return Empty() |
||||
|
||||
def getLastAsaUpload( self, request, context ): |
||||
"""Get the last ASL Scenario Archive upload.""" |
||||
last_asa_upload = webapp_scenarios._last_asa_upload #pylint: disable=protected-access |
||||
# return the last ASL Scenario Archive upload |
||||
_logger.debug( "- Returning the last ASA upload: %s", last_asa_upload ) |
||||
if last_asa_upload: |
||||
for key in ("vasl_setup","screenshot"): |
||||
if last_asa_upload.get( key ): |
||||
last_asa_upload[key] = base64.b64encode( last_asa_upload[key] ).decode( "ascii" ) |
||||
return GetLastAsaUploadResponse( lastUploadJson = json.dumps( last_asa_upload ) ) |
||||
|
||||
def dumpVsav( self, request, context ): |
||||
"""Dump a VASL save file.""" |
||||
self._log_request( request, context ) |
||||
# dump the VSAV |
||||
with TempFile( mode="wb" ) as temp_file: |
||||
temp_file.write( request.vsavData ) |
||||
temp_file.close( delete=False ) |
||||
vassal_shim = VassalShim() |
||||
vsav_dump = vassal_shim.dump_scenario( temp_file.name ) |
||||
_logger.debug( "- VSAV dump: #bytes=%s", len(vsav_dump) ) |
||||
return DumpVsavResponse( vsavDump=vsav_dump ) |
||||
|
||||
def getVaslPieces( self, request, context ): |
||||
"""Get the pieces for the specified VASL module.""" |
||||
self._log_request( request, context ) |
||||
vasl_version = request.vaslVersion |
||||
|
||||
# dump the VASL pieces |
||||
fname = self._vasl_mods[ vasl_version ] |
||||
vasl_mod = webapp_vasl_mod.VaslMod( fname, self._webapp.config["DATA_DIR"], None ) |
||||
buf = io.StringIO() |
||||
results = [ [ "GPID", "Name", "Front images", "Back images"] ] |
||||
pieces = vasl_mod._pieces #pylint: disable=protected-access |
||||
# GPID's were originally int's but then changed to str's. We then started seeing non-numeric GPID's :-/ |
||||
# For back-compat, we try to maintain sort order for numeric values. |
||||
def sort_key( val ): #pylint: disable=missing-docstring |
||||
if val.isdigit(): |
||||
return ( "0"*10 + val )[-10:] |
||||
else: |
||||
# make sure that alphanumeric values appear after numeric values, even if they start with a number |
||||
return "_" + val |
||||
gpids = sorted( pieces.keys(), key=sort_key ) # nb: because GPID's changed from int to str :-/ |
||||
for gpid in gpids: |
||||
piece = pieces[ gpid ] |
||||
assert piece["gpid"] == gpid |
||||
results.append( [ gpid, piece["name"], piece["front_images"], piece["back_images"] ] ) |
||||
print( tabulate.tabulate( results, headers="firstrow", numalign="left" ), file=buf ) |
||||
|
||||
# get the piece GPID's |
||||
gpids = webapp_vasl_mod.get_vo_gpids( vasl_mod ) |
||||
|
||||
return GetVaslPiecesResponse( |
||||
pieceDump=buf.getvalue(), gpids=gpids |
||||
) |
||||
|
||||
def getAppConfig( self, request, context ): |
||||
"""Get the app config.""" |
||||
self._log_request( request, context ) |
||||
# get the app config |
||||
app_config = self._webapp.config |
||||
_logger.debug( "- %s", app_config ) |
||||
return GetAppConfigResponse( |
||||
appConfigJson = json.dumps( app_config, default=str ) |
||||
) |
||||
|
||||
def setAppConfigVal( self, request, context ): |
||||
"""Set an app config value.""" |
||||
self._log_request( request, context ) |
||||
# get the app config setting |
||||
for val_type in ( "strVal", "intVal", "boolVal" ): |
||||
if request.HasField( val_type ): |
||||
key, val = request.key, getattr(request,val_type) |
||||
_logger.debug( "- Setting app config: %s = %s (%s)", key, str(val), type(val).__name__ ) |
||||
self._webapp.config[ key ] = val |
||||
return Empty() |
||||
raise RuntimeError( "Can't find app config key." ) |
||||
|
||||
def deleteAppConfigVal( self, request, context ): |
||||
"""Delete an app config value.""" |
||||
self._log_request( request, context ) |
||||
key = request.key |
||||
# delete the app config setting |
||||
_logger.debug( "- Deleting app config: %s", key ) |
||||
if key in self._webapp.config: |
||||
del self._webapp.config[ key ] |
||||
return Empty() |
||||
|
||||
def saveTempFile( self, request, context ): |
||||
"""Save a temp file.""" |
||||
self._log_request( request, context ) |
||||
fname, data = request.fileName, request.data |
||||
# save the temp file |
||||
fname = os.path.join( self._temp_dir.name, fname ) |
||||
_logger.debug( "- Saving temp file (#bytes=%d): %s", len(data), fname ) |
||||
with open( fname, "wb" ) as fp: |
||||
fp.write( data ) |
||||
return Empty() |
||||
|
||||
@staticmethod |
||||
def _log_request( req, ctx ): |
||||
"""Log a request.""" |
||||
if ctx is None: |
||||
return # nb: we don't log internal calls |
||||
# get the entry-point name |
||||
msg = "{}()".format( inspect.currentframe().f_back.f_code.co_name ) |
||||
# add the brief request info |
||||
func = getattr( req, "brief", None ) |
||||
if func: |
||||
brief = func() |
||||
if brief: |
||||
msg += ": {}".format( brief ) |
||||
# add the request dump |
||||
func = getattr( req, "dump", None ) |
||||
if func: |
||||
buf = io.StringIO() |
||||
func( out=buf ) |
||||
buf = buf.getvalue().strip() |
||||
if buf: |
||||
msg += "\n{}".format( buf ) |
||||
# log the message |
||||
_logger.info( "TEST CONTROL: %s", msg ) |
||||
Binary file not shown.
Binary file not shown.
@ -0,0 +1,32 @@ |
||||
"""gRPC protobuf definitions (for controlling tests).""" |
||||
|
||||
import sys |
||||
import importlib |
||||
|
||||
# --------------------------------------------------------------------- |
||||
|
||||
def _init_classes(): |
||||
"""Initialize the gRPC classes.""" |
||||
|
||||
# process each request/response class |
||||
from .utils import get_classes, split_words |
||||
for cls in get_classes(): |
||||
|
||||
# check if the class has a corresponding module |
||||
words = split_words( cls.__name__ ) |
||||
mod_name = "_".join( words ) |
||||
try: |
||||
mod2 = importlib.import_module( "vasl_templates.webapp.tests.proto." + mod_name ) |
||||
except ModuleNotFoundError: |
||||
continue |
||||
|
||||
# yup - inject the functions into the class |
||||
for elem2 in dir(mod2): |
||||
obj = getattr( mod2, elem2 ) |
||||
if not callable( obj ): |
||||
continue |
||||
setattr( cls, elem2, obj ) |
||||
|
||||
# --------------------------------------------------------------------- |
||||
|
||||
_init_classes() |
||||
@ -0,0 +1,164 @@ |
||||
syntax = "proto3" ; |
||||
|
||||
import "google/protobuf/empty.proto" ; |
||||
|
||||
// -------------------------------------------------------------------- |
||||
|
||||
message StartTestsResponse { |
||||
repeated string capabilities = 1 ; |
||||
} |
||||
|
||||
// - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - |
||||
|
||||
message GetVassalVersionsResponse { |
||||
repeated string vassalVersions = 1 ; |
||||
} |
||||
|
||||
message SetVassalVersionRequest { |
||||
string vassalVersion = 1 ; |
||||
} |
||||
|
||||
message GetVaslVersionsResponse { |
||||
repeated string vaslVersions = 1 ; |
||||
} |
||||
|
||||
message SetVaslVersionRequest { |
||||
enum VaslExtnsType { NONE=0 ; REAL=1 ; TEMP_DIR=2 ; } |
||||
string vaslVersion = 1 ; |
||||
VaslExtnsType vaslExtnsType = 2 ; |
||||
} |
||||
|
||||
message GetVaslExtnsResponse { |
||||
string vaslExtnsJson = 1 ; |
||||
} |
||||
|
||||
message SetVaslExtnInfoDirRequest { |
||||
string dirName = 1 ; // nb: relative to the fixtures directory |
||||
} |
||||
|
||||
message SetGpidRemappingsRequest { |
||||
string gpidRemappingsJson = 1 ; |
||||
} |
||||
|
||||
message GetVaslModWarningsResponse { |
||||
repeated string warnings = 1 ; |
||||
} |
||||
|
||||
// - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - |
||||
|
||||
message SetDataDirRequest { |
||||
enum DirType { TEST=0 ; REAL=1 ; } |
||||
DirType dirType = 1 ; |
||||
} |
||||
|
||||
message SetDefaultScenarioRequest { |
||||
string fileName = 1 ; // nb: relative to the fixtures directory |
||||
} |
||||
|
||||
message SetDefaultTemplatePackRequest { |
||||
enum TemplatePackType { DEFAULT=0 ; REAL=1 ; } |
||||
oneof tp_oneof { |
||||
TemplatePackType templatePackType = 1 ; |
||||
string dirName = 2 ; // nb: relative to the fixtures directory |
||||
bytes zipData = 3 ; |
||||
} |
||||
} |
||||
|
||||
message SetVehOrdNotesDirRequest { |
||||
enum DirType { NONE=0 ; TEST=1 ; REAL=2 ; } |
||||
DirType dirType = 1 ; |
||||
} |
||||
|
||||
message SetUserFilesDirRequest { |
||||
string dirOrUrl = 1 ; // nb: relative to the fixtures directory |
||||
} |
||||
|
||||
message SetAsaScenarioIndexRequest { |
||||
string fileName = 1 ; // nb: relative to the fixtures directory |
||||
} |
||||
|
||||
message SetRoarScenarioIndexRequest { |
||||
string fileName = 1 ; // nb: relative to the fixtures directory |
||||
} |
||||
|
||||
message GetLastSnippetImageResponse { |
||||
bytes imageData = 1 ; |
||||
} |
||||
|
||||
message GetLastAsaUploadResponse { |
||||
string lastUploadJson = 1 ; |
||||
} |
||||
|
||||
// - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - |
||||
|
||||
message DumpVsavRequest { |
||||
bytes vsavData = 1 ; |
||||
} |
||||
message DumpVsavResponse { |
||||
string vsavDump = 1 ; |
||||
} |
||||
|
||||
message GetVaslPiecesRequest { |
||||
string vaslVersion = 1 ; |
||||
} |
||||
message GetVaslPiecesResponse { |
||||
string pieceDump = 1 ; |
||||
repeated string gpids = 2 ; |
||||
} |
||||
|
||||
message GetAppConfigResponse { |
||||
string appConfigJson = 1 ; |
||||
} |
||||
|
||||
message SetAppConfigValRequest { |
||||
string key = 1 ; |
||||
oneof ac_oneof { |
||||
string strVal = 2 ; |
||||
int32 intVal = 3 ; |
||||
bool boolVal = 4 ; |
||||
} |
||||
} |
||||
|
||||
message DeleteAppConfigValRequest { |
||||
string key = 1 ; |
||||
} |
||||
|
||||
message SaveTempFileRequest { |
||||
string fileName = 1 ; // nb: relative to the servicer's temp directory |
||||
bytes data = 2 ; |
||||
} |
||||
|
||||
// -------------------------------------------------------------------- |
||||
|
||||
service ControlTests |
||||
{ |
||||
rpc startTests( google.protobuf.Empty ) returns ( StartTestsResponse ) ; |
||||
rpc endTests( google.protobuf.Empty ) returns ( google.protobuf.Empty ) ; |
||||
|
||||
rpc getVassalVersions( google.protobuf.Empty ) returns ( GetVassalVersionsResponse ) ; |
||||
rpc setVassalVersion( SetVassalVersionRequest ) returns ( google.protobuf.Empty ) ; |
||||
rpc getVaslVersions( google.protobuf.Empty ) returns ( GetVaslVersionsResponse ) ; |
||||
rpc setVaslVersion( SetVaslVersionRequest ) returns ( google.protobuf.Empty ) ; |
||||
rpc getVaslExtns( google.protobuf.Empty ) returns ( GetVaslExtnsResponse ) ; |
||||
rpc setVaslExtnInfoDir( SetVaslExtnInfoDirRequest ) returns ( google.protobuf.Empty ) ; |
||||
rpc setGpidRemappings( SetGpidRemappingsRequest ) returns ( google.protobuf.Empty ) ; |
||||
rpc getVaslModWarnings( google.protobuf.Empty ) returns ( GetVaslModWarningsResponse ) ; |
||||
|
||||
rpc setDataDir( SetDataDirRequest ) returns ( google.protobuf.Empty ) ; |
||||
rpc setDefaultScenario( SetDefaultScenarioRequest ) returns ( google.protobuf.Empty ) ; |
||||
rpc setDefaultTemplatePack( SetDefaultTemplatePackRequest ) returns ( google.protobuf.Empty ) ; |
||||
rpc setVehOrdNotesDir( SetVehOrdNotesDirRequest ) returns ( google.protobuf.Empty ) ; |
||||
rpc setUserFilesDir( SetUserFilesDirRequest ) returns ( google.protobuf.Empty ) ; |
||||
rpc setAsaScenarioIndex( SetAsaScenarioIndexRequest ) returns ( google.protobuf.Empty ) ; |
||||
rpc setRoarScenarioIndex( SetRoarScenarioIndexRequest ) returns ( google.protobuf.Empty ) ; |
||||
rpc getLastSnippetImage( google.protobuf.Empty ) returns ( GetLastSnippetImageResponse ) ; |
||||
rpc resetLastAsaUpload( google.protobuf.Empty ) returns ( google.protobuf.Empty ) ; |
||||
rpc getLastAsaUpload( google.protobuf.Empty ) returns ( GetLastAsaUploadResponse ) ; |
||||
|
||||
rpc dumpVsav( DumpVsavRequest ) returns ( DumpVsavResponse ) ; |
||||
rpc getVaslPieces( GetVaslPiecesRequest ) returns ( GetVaslPiecesResponse ) ; |
||||
rpc getAppConfig( google.protobuf.Empty ) returns ( GetAppConfigResponse ) ; |
||||
rpc setAppConfigVal( SetAppConfigValRequest ) returns ( google.protobuf.Empty ) ; |
||||
rpc deleteAppConfigVal( DeleteAppConfigValRequest ) returns ( google.protobuf.Empty ) ; |
||||
rpc saveTempFile( SaveTempFileRequest ) returns ( google.protobuf.Empty ) ; |
||||
} |
||||
@ -0,0 +1,7 @@ |
||||
""" Injected functions for DeleteAppConfigValRequest. """ |
||||
|
||||
# --------------------------------------------------------------------- |
||||
|
||||
def brief( self ): |
||||
"""Return a DeleteAppConfigValRequest as a brief string.""" |
||||
return self.key |
||||
@ -0,0 +1,7 @@ |
||||
""" Injected functions for DumpVsavRequest. """ |
||||
|
||||
# --------------------------------------------------------------------- |
||||
|
||||
def brief( self ): |
||||
"""Return a DumpVsavRequest as a brief string.""" |
||||
return "#bytes={}".format( len(self.vsavData) ) |
||||
File diff suppressed because one or more lines are too long
@ -0,0 +1,898 @@ |
||||
# Generated by the gRPC Python protocol compiler plugin. DO NOT EDIT! |
||||
"""Client and server classes corresponding to protobuf-defined services.""" |
||||
import grpc |
||||
|
||||
import vasl_templates.webapp.tests.proto.generated.control_tests_pb2 as control__tests__pb2 |
||||
from google.protobuf import empty_pb2 as google_dot_protobuf_dot_empty__pb2 |
||||
|
||||
|
||||
class ControlTestsStub(object): |
||||
"""-------------------------------------------------------------------- |
||||
|
||||
""" |
||||
|
||||
def __init__(self, channel): |
||||
"""Constructor. |
||||
|
||||
Args: |
||||
channel: A grpc.Channel. |
||||
""" |
||||
self.startTests = channel.unary_unary( |
||||
'/ControlTests/startTests', |
||||
request_serializer=google_dot_protobuf_dot_empty__pb2.Empty.SerializeToString, |
||||
response_deserializer=control__tests__pb2.StartTestsResponse.FromString, |
||||
) |
||||
self.endTests = channel.unary_unary( |
||||
'/ControlTests/endTests', |
||||
request_serializer=google_dot_protobuf_dot_empty__pb2.Empty.SerializeToString, |
||||
response_deserializer=google_dot_protobuf_dot_empty__pb2.Empty.FromString, |
||||
) |
||||
self.getVassalVersions = channel.unary_unary( |
||||
'/ControlTests/getVassalVersions', |
||||
request_serializer=google_dot_protobuf_dot_empty__pb2.Empty.SerializeToString, |
||||
response_deserializer=control__tests__pb2.GetVassalVersionsResponse.FromString, |
||||
) |
||||
self.setVassalVersion = channel.unary_unary( |
||||
'/ControlTests/setVassalVersion', |
||||
request_serializer=control__tests__pb2.SetVassalVersionRequest.SerializeToString, |
||||
response_deserializer=google_dot_protobuf_dot_empty__pb2.Empty.FromString, |
||||
) |
||||
self.getVaslVersions = channel.unary_unary( |
||||
'/ControlTests/getVaslVersions', |
||||
request_serializer=google_dot_protobuf_dot_empty__pb2.Empty.SerializeToString, |
||||
response_deserializer=control__tests__pb2.GetVaslVersionsResponse.FromString, |
||||
) |
||||
self.setVaslVersion = channel.unary_unary( |
||||
'/ControlTests/setVaslVersion', |
||||
request_serializer=control__tests__pb2.SetVaslVersionRequest.SerializeToString, |
||||
response_deserializer=google_dot_protobuf_dot_empty__pb2.Empty.FromString, |
||||
) |
||||
self.getVaslExtns = channel.unary_unary( |
||||
'/ControlTests/getVaslExtns', |
||||
request_serializer=google_dot_protobuf_dot_empty__pb2.Empty.SerializeToString, |
||||
response_deserializer=control__tests__pb2.GetVaslExtnsResponse.FromString, |
||||
) |
||||
self.setVaslExtnInfoDir = channel.unary_unary( |
||||
'/ControlTests/setVaslExtnInfoDir', |
||||
request_serializer=control__tests__pb2.SetVaslExtnInfoDirRequest.SerializeToString, |
||||
response_deserializer=google_dot_protobuf_dot_empty__pb2.Empty.FromString, |
||||
) |
||||
self.setGpidRemappings = channel.unary_unary( |
||||
'/ControlTests/setGpidRemappings', |
||||
request_serializer=control__tests__pb2.SetGpidRemappingsRequest.SerializeToString, |
||||
response_deserializer=google_dot_protobuf_dot_empty__pb2.Empty.FromString, |
||||
) |
||||
self.getVaslModWarnings = channel.unary_unary( |
||||
'/ControlTests/getVaslModWarnings', |
||||
request_serializer=google_dot_protobuf_dot_empty__pb2.Empty.SerializeToString, |
||||
response_deserializer=control__tests__pb2.GetVaslModWarningsResponse.FromString, |
||||
) |
||||
self.setDataDir = channel.unary_unary( |
||||
'/ControlTests/setDataDir', |
||||
request_serializer=control__tests__pb2.SetDataDirRequest.SerializeToString, |
||||
response_deserializer=google_dot_protobuf_dot_empty__pb2.Empty.FromString, |
||||
) |
||||
self.setDefaultScenario = channel.unary_unary( |
||||
'/ControlTests/setDefaultScenario', |
||||
request_serializer=control__tests__pb2.SetDefaultScenarioRequest.SerializeToString, |
||||
response_deserializer=google_dot_protobuf_dot_empty__pb2.Empty.FromString, |
||||
) |
||||
self.setDefaultTemplatePack = channel.unary_unary( |
||||
'/ControlTests/setDefaultTemplatePack', |
||||
request_serializer=control__tests__pb2.SetDefaultTemplatePackRequest.SerializeToString, |
||||
response_deserializer=google_dot_protobuf_dot_empty__pb2.Empty.FromString, |
||||
) |
||||
self.setVehOrdNotesDir = channel.unary_unary( |
||||
'/ControlTests/setVehOrdNotesDir', |
||||
request_serializer=control__tests__pb2.SetVehOrdNotesDirRequest.SerializeToString, |
||||
response_deserializer=google_dot_protobuf_dot_empty__pb2.Empty.FromString, |
||||
) |
||||
self.setUserFilesDir = channel.unary_unary( |
||||
'/ControlTests/setUserFilesDir', |
||||
request_serializer=control__tests__pb2.SetUserFilesDirRequest.SerializeToString, |
||||
response_deserializer=google_dot_protobuf_dot_empty__pb2.Empty.FromString, |
||||
) |
||||
self.setAsaScenarioIndex = channel.unary_unary( |
||||
'/ControlTests/setAsaScenarioIndex', |
||||
request_serializer=control__tests__pb2.SetAsaScenarioIndexRequest.SerializeToString, |
||||
response_deserializer=google_dot_protobuf_dot_empty__pb2.Empty.FromString, |
||||
) |
||||
self.setRoarScenarioIndex = channel.unary_unary( |
||||
'/ControlTests/setRoarScenarioIndex', |
||||
request_serializer=control__tests__pb2.SetRoarScenarioIndexRequest.SerializeToString, |
||||
response_deserializer=google_dot_protobuf_dot_empty__pb2.Empty.FromString, |
||||
) |
||||
self.getLastSnippetImage = channel.unary_unary( |
||||
'/ControlTests/getLastSnippetImage', |
||||
request_serializer=google_dot_protobuf_dot_empty__pb2.Empty.SerializeToString, |
||||
response_deserializer=control__tests__pb2.GetLastSnippetImageResponse.FromString, |
||||
) |
||||
self.resetLastAsaUpload = channel.unary_unary( |
||||
'/ControlTests/resetLastAsaUpload', |
||||
request_serializer=google_dot_protobuf_dot_empty__pb2.Empty.SerializeToString, |
||||
response_deserializer=google_dot_protobuf_dot_empty__pb2.Empty.FromString, |
||||
) |
||||
self.getLastAsaUpload = channel.unary_unary( |
||||
'/ControlTests/getLastAsaUpload', |
||||
request_serializer=google_dot_protobuf_dot_empty__pb2.Empty.SerializeToString, |
||||
response_deserializer=control__tests__pb2.GetLastAsaUploadResponse.FromString, |
||||
) |
||||
self.dumpVsav = channel.unary_unary( |
||||
'/ControlTests/dumpVsav', |
||||
request_serializer=control__tests__pb2.DumpVsavRequest.SerializeToString, |
||||
response_deserializer=control__tests__pb2.DumpVsavResponse.FromString, |
||||
) |
||||
self.getVaslPieces = channel.unary_unary( |
||||
'/ControlTests/getVaslPieces', |
||||
request_serializer=control__tests__pb2.GetVaslPiecesRequest.SerializeToString, |
||||
response_deserializer=control__tests__pb2.GetVaslPiecesResponse.FromString, |
||||
) |
||||
self.getAppConfig = channel.unary_unary( |
||||
'/ControlTests/getAppConfig', |
||||
request_serializer=google_dot_protobuf_dot_empty__pb2.Empty.SerializeToString, |
||||
response_deserializer=control__tests__pb2.GetAppConfigResponse.FromString, |
||||
) |
||||
self.setAppConfigVal = channel.unary_unary( |
||||
'/ControlTests/setAppConfigVal', |
||||
request_serializer=control__tests__pb2.SetAppConfigValRequest.SerializeToString, |
||||
response_deserializer=google_dot_protobuf_dot_empty__pb2.Empty.FromString, |
||||
) |
||||
self.deleteAppConfigVal = channel.unary_unary( |
||||
'/ControlTests/deleteAppConfigVal', |
||||
request_serializer=control__tests__pb2.DeleteAppConfigValRequest.SerializeToString, |
||||
response_deserializer=google_dot_protobuf_dot_empty__pb2.Empty.FromString, |
||||
) |
||||
self.saveTempFile = channel.unary_unary( |
||||
'/ControlTests/saveTempFile', |
||||
request_serializer=control__tests__pb2.SaveTempFileRequest.SerializeToString, |
||||
response_deserializer=google_dot_protobuf_dot_empty__pb2.Empty.FromString, |
||||
) |
||||
|
||||
|
||||
class ControlTestsServicer(object): |
||||
"""-------------------------------------------------------------------- |
||||
|
||||
""" |
||||
|
||||
def startTests(self, request, context): |
||||
"""Missing associated documentation comment in .proto file.""" |
||||
context.set_code(grpc.StatusCode.UNIMPLEMENTED) |
||||
context.set_details('Method not implemented!') |
||||
raise NotImplementedError('Method not implemented!') |
||||
|
||||
def endTests(self, request, context): |
||||
"""Missing associated documentation comment in .proto file.""" |
||||
context.set_code(grpc.StatusCode.UNIMPLEMENTED) |
||||
context.set_details('Method not implemented!') |
||||
raise NotImplementedError('Method not implemented!') |
||||
|
||||
def getVassalVersions(self, request, context): |
||||
"""Missing associated documentation comment in .proto file.""" |
||||
context.set_code(grpc.StatusCode.UNIMPLEMENTED) |
||||
context.set_details('Method not implemented!') |
||||
raise NotImplementedError('Method not implemented!') |
||||
|
||||
def setVassalVersion(self, request, context): |
||||
"""Missing associated documentation comment in .proto file.""" |
||||
context.set_code(grpc.StatusCode.UNIMPLEMENTED) |
||||
context.set_details('Method not implemented!') |
||||
raise NotImplementedError('Method not implemented!') |
||||
|
||||
def getVaslVersions(self, request, context): |
||||
"""Missing associated documentation comment in .proto file.""" |
||||
context.set_code(grpc.StatusCode.UNIMPLEMENTED) |
||||
context.set_details('Method not implemented!') |
||||
raise NotImplementedError('Method not implemented!') |
||||
|
||||
def setVaslVersion(self, request, context): |
||||
"""Missing associated documentation comment in .proto file.""" |
||||
context.set_code(grpc.StatusCode.UNIMPLEMENTED) |
||||
context.set_details('Method not implemented!') |
||||
raise NotImplementedError('Method not implemented!') |
||||
|
||||
def getVaslExtns(self, request, context): |
||||
"""Missing associated documentation comment in .proto file.""" |
||||
context.set_code(grpc.StatusCode.UNIMPLEMENTED) |
||||
context.set_details('Method not implemented!') |
||||
raise NotImplementedError('Method not implemented!') |
||||
|
||||
def setVaslExtnInfoDir(self, request, context): |
||||
"""Missing associated documentation comment in .proto file.""" |
||||
context.set_code(grpc.StatusCode.UNIMPLEMENTED) |
||||
context.set_details('Method not implemented!') |
||||
raise NotImplementedError('Method not implemented!') |
||||
|
||||
def setGpidRemappings(self, request, context): |
||||
"""Missing associated documentation comment in .proto file.""" |
||||
context.set_code(grpc.StatusCode.UNIMPLEMENTED) |
||||
context.set_details('Method not implemented!') |
||||
raise NotImplementedError('Method not implemented!') |
||||
|
||||
def getVaslModWarnings(self, request, context): |
||||
"""Missing associated documentation comment in .proto file.""" |
||||
context.set_code(grpc.StatusCode.UNIMPLEMENTED) |
||||
context.set_details('Method not implemented!') |
||||
raise NotImplementedError('Method not implemented!') |
||||
|
||||
def setDataDir(self, request, context): |
||||
"""Missing associated documentation comment in .proto file.""" |
||||
context.set_code(grpc.StatusCode.UNIMPLEMENTED) |
||||
context.set_details('Method not implemented!') |
||||
raise NotImplementedError('Method not implemented!') |
||||
|
||||
def setDefaultScenario(self, request, context): |
||||
"""Missing associated documentation comment in .proto file.""" |
||||
context.set_code(grpc.StatusCode.UNIMPLEMENTED) |
||||
context.set_details('Method not implemented!') |
||||
raise NotImplementedError('Method not implemented!') |
||||
|
||||
def setDefaultTemplatePack(self, request, context): |
||||
"""Missing associated documentation comment in .proto file.""" |
||||
context.set_code(grpc.StatusCode.UNIMPLEMENTED) |
||||
context.set_details('Method not implemented!') |
||||
raise NotImplementedError('Method not implemented!') |
||||
|
||||
def setVehOrdNotesDir(self, request, context): |
||||
"""Missing associated documentation comment in .proto file.""" |
||||
context.set_code(grpc.StatusCode.UNIMPLEMENTED) |
||||
context.set_details('Method not implemented!') |
||||
raise NotImplementedError('Method not implemented!') |
||||
|
||||
def setUserFilesDir(self, request, context): |
||||
"""Missing associated documentation comment in .proto file.""" |
||||
context.set_code(grpc.StatusCode.UNIMPLEMENTED) |
||||
context.set_details('Method not implemented!') |
||||
raise NotImplementedError('Method not implemented!') |
||||
|
||||
def setAsaScenarioIndex(self, request, context): |
||||
"""Missing associated documentation comment in .proto file.""" |
||||
context.set_code(grpc.StatusCode.UNIMPLEMENTED) |
||||
context.set_details('Method not implemented!') |
||||
raise NotImplementedError('Method not implemented!') |
||||
|
||||
def setRoarScenarioIndex(self, request, context): |
||||
"""Missing associated documentation comment in .proto file.""" |
||||
context.set_code(grpc.StatusCode.UNIMPLEMENTED) |
||||
context.set_details('Method not implemented!') |
||||
raise NotImplementedError('Method not implemented!') |
||||
|
||||
def getLastSnippetImage(self, request, context): |
||||
"""Missing associated documentation comment in .proto file.""" |
||||
context.set_code(grpc.StatusCode.UNIMPLEMENTED) |
||||
context.set_details('Method not implemented!') |
||||
raise NotImplementedError('Method not implemented!') |
||||
|
||||
def resetLastAsaUpload(self, request, context): |
||||
"""Missing associated documentation comment in .proto file.""" |
||||
context.set_code(grpc.StatusCode.UNIMPLEMENTED) |
||||
context.set_details('Method not implemented!') |
||||
raise NotImplementedError('Method not implemented!') |
||||
|
||||
def getLastAsaUpload(self, request, context): |
||||
"""Missing associated documentation comment in .proto file.""" |
||||
context.set_code(grpc.StatusCode.UNIMPLEMENTED) |
||||
context.set_details('Method not implemented!') |
||||
raise NotImplementedError('Method not implemented!') |
||||
|
||||
def dumpVsav(self, request, context): |
||||
"""Missing associated documentation comment in .proto file.""" |
||||
context.set_code(grpc.StatusCode.UNIMPLEMENTED) |
||||
context.set_details('Method not implemented!') |
||||
raise NotImplementedError('Method not implemented!') |
||||
|
||||
def getVaslPieces(self, request, context): |
||||
"""Missing associated documentation comment in .proto file.""" |
||||
context.set_code(grpc.StatusCode.UNIMPLEMENTED) |
||||
context.set_details('Method not implemented!') |
||||
raise NotImplementedError('Method not implemented!') |
||||
|
||||
def getAppConfig(self, request, context): |
||||
"""Missing associated documentation comment in .proto file.""" |
||||
context.set_code(grpc.StatusCode.UNIMPLEMENTED) |
||||
context.set_details('Method not implemented!') |
||||
raise NotImplementedError('Method not implemented!') |
||||
|
||||
def setAppConfigVal(self, request, context): |
||||
"""Missing associated documentation comment in .proto file.""" |
||||
context.set_code(grpc.StatusCode.UNIMPLEMENTED) |
||||
context.set_details('Method not implemented!') |
||||
raise NotImplementedError('Method not implemented!') |
||||
|
||||
def deleteAppConfigVal(self, request, context): |
||||
"""Missing associated documentation comment in .proto file.""" |
||||
context.set_code(grpc.StatusCode.UNIMPLEMENTED) |
||||
context.set_details('Method not implemented!') |
||||
raise NotImplementedError('Method not implemented!') |
||||
|
||||
def saveTempFile(self, request, context): |
||||
"""Missing associated documentation comment in .proto file.""" |
||||
context.set_code(grpc.StatusCode.UNIMPLEMENTED) |
||||
context.set_details('Method not implemented!') |
||||
raise NotImplementedError('Method not implemented!') |
||||
|
||||
|
||||
def add_ControlTestsServicer_to_server(servicer, server): |
||||
rpc_method_handlers = { |
||||
'startTests': grpc.unary_unary_rpc_method_handler( |
||||
servicer.startTests, |
||||
request_deserializer=google_dot_protobuf_dot_empty__pb2.Empty.FromString, |
||||
response_serializer=control__tests__pb2.StartTestsResponse.SerializeToString, |
||||
), |
||||
'endTests': grpc.unary_unary_rpc_method_handler( |
||||
servicer.endTests, |
||||
request_deserializer=google_dot_protobuf_dot_empty__pb2.Empty.FromString, |
||||
response_serializer=google_dot_protobuf_dot_empty__pb2.Empty.SerializeToString, |
||||
), |
||||
'getVassalVersions': grpc.unary_unary_rpc_method_handler( |
||||
servicer.getVassalVersions, |
||||
request_deserializer=google_dot_protobuf_dot_empty__pb2.Empty.FromString, |
||||
response_serializer=control__tests__pb2.GetVassalVersionsResponse.SerializeToString, |
||||
), |
||||
'setVassalVersion': grpc.unary_unary_rpc_method_handler( |
||||
servicer.setVassalVersion, |
||||
request_deserializer=control__tests__pb2.SetVassalVersionRequest.FromString, |
||||
response_serializer=google_dot_protobuf_dot_empty__pb2.Empty.SerializeToString, |
||||
), |
||||
'getVaslVersions': grpc.unary_unary_rpc_method_handler( |
||||
servicer.getVaslVersions, |
||||
request_deserializer=google_dot_protobuf_dot_empty__pb2.Empty.FromString, |
||||
response_serializer=control__tests__pb2.GetVaslVersionsResponse.SerializeToString, |
||||
), |
||||
'setVaslVersion': grpc.unary_unary_rpc_method_handler( |
||||
servicer.setVaslVersion, |
||||
request_deserializer=control__tests__pb2.SetVaslVersionRequest.FromString, |
||||
response_serializer=google_dot_protobuf_dot_empty__pb2.Empty.SerializeToString, |
||||
), |
||||
'getVaslExtns': grpc.unary_unary_rpc_method_handler( |
||||
servicer.getVaslExtns, |
||||
request_deserializer=google_dot_protobuf_dot_empty__pb2.Empty.FromString, |
||||
response_serializer=control__tests__pb2.GetVaslExtnsResponse.SerializeToString, |
||||
), |
||||
'setVaslExtnInfoDir': grpc.unary_unary_rpc_method_handler( |
||||
servicer.setVaslExtnInfoDir, |
||||
request_deserializer=control__tests__pb2.SetVaslExtnInfoDirRequest.FromString, |
||||
response_serializer=google_dot_protobuf_dot_empty__pb2.Empty.SerializeToString, |
||||
), |
||||
'setGpidRemappings': grpc.unary_unary_rpc_method_handler( |
||||
servicer.setGpidRemappings, |
||||
request_deserializer=control__tests__pb2.SetGpidRemappingsRequest.FromString, |
||||
response_serializer=google_dot_protobuf_dot_empty__pb2.Empty.SerializeToString, |
||||
), |
||||
'getVaslModWarnings': grpc.unary_unary_rpc_method_handler( |
||||
servicer.getVaslModWarnings, |
||||
request_deserializer=google_dot_protobuf_dot_empty__pb2.Empty.FromString, |
||||
response_serializer=control__tests__pb2.GetVaslModWarningsResponse.SerializeToString, |
||||
), |
||||
'setDataDir': grpc.unary_unary_rpc_method_handler( |
||||
servicer.setDataDir, |
||||
request_deserializer=control__tests__pb2.SetDataDirRequest.FromString, |
||||
response_serializer=google_dot_protobuf_dot_empty__pb2.Empty.SerializeToString, |
||||
), |
||||
'setDefaultScenario': grpc.unary_unary_rpc_method_handler( |
||||
servicer.setDefaultScenario, |
||||
request_deserializer=control__tests__pb2.SetDefaultScenarioRequest.FromString, |
||||
response_serializer=google_dot_protobuf_dot_empty__pb2.Empty.SerializeToString, |
||||
), |
||||
'setDefaultTemplatePack': grpc.unary_unary_rpc_method_handler( |
||||
servicer.setDefaultTemplatePack, |
||||
request_deserializer=control__tests__pb2.SetDefaultTemplatePackRequest.FromString, |
||||
response_serializer=google_dot_protobuf_dot_empty__pb2.Empty.SerializeToString, |
||||
), |
||||
'setVehOrdNotesDir': grpc.unary_unary_rpc_method_handler( |
||||
servicer.setVehOrdNotesDir, |
||||
request_deserializer=control__tests__pb2.SetVehOrdNotesDirRequest.FromString, |
||||
response_serializer=google_dot_protobuf_dot_empty__pb2.Empty.SerializeToString, |
||||
), |
||||
'setUserFilesDir': grpc.unary_unary_rpc_method_handler( |
||||
servicer.setUserFilesDir, |
||||
request_deserializer=control__tests__pb2.SetUserFilesDirRequest.FromString, |
||||
response_serializer=google_dot_protobuf_dot_empty__pb2.Empty.SerializeToString, |
||||
), |
||||
'setAsaScenarioIndex': grpc.unary_unary_rpc_method_handler( |
||||
servicer.setAsaScenarioIndex, |
||||
request_deserializer=control__tests__pb2.SetAsaScenarioIndexRequest.FromString, |
||||
response_serializer=google_dot_protobuf_dot_empty__pb2.Empty.SerializeToString, |
||||
), |
||||
'setRoarScenarioIndex': grpc.unary_unary_rpc_method_handler( |
||||
servicer.setRoarScenarioIndex, |
||||
request_deserializer=control__tests__pb2.SetRoarScenarioIndexRequest.FromString, |
||||
response_serializer=google_dot_protobuf_dot_empty__pb2.Empty.SerializeToString, |
||||
), |
||||
'getLastSnippetImage': grpc.unary_unary_rpc_method_handler( |
||||
servicer.getLastSnippetImage, |
||||
request_deserializer=google_dot_protobuf_dot_empty__pb2.Empty.FromString, |
||||
response_serializer=control__tests__pb2.GetLastSnippetImageResponse.SerializeToString, |
||||
), |
||||
'resetLastAsaUpload': grpc.unary_unary_rpc_method_handler( |
||||
servicer.resetLastAsaUpload, |
||||
request_deserializer=google_dot_protobuf_dot_empty__pb2.Empty.FromString, |
||||
response_serializer=google_dot_protobuf_dot_empty__pb2.Empty.SerializeToString, |
||||
), |
||||
'getLastAsaUpload': grpc.unary_unary_rpc_method_handler( |
||||
servicer.getLastAsaUpload, |
||||
request_deserializer=google_dot_protobuf_dot_empty__pb2.Empty.FromString, |
||||
response_serializer=control__tests__pb2.GetLastAsaUploadResponse.SerializeToString, |
||||
), |
||||
'dumpVsav': grpc.unary_unary_rpc_method_handler( |
||||
servicer.dumpVsav, |
||||
request_deserializer=control__tests__pb2.DumpVsavRequest.FromString, |
||||
response_serializer=control__tests__pb2.DumpVsavResponse.SerializeToString, |
||||
), |
||||
'getVaslPieces': grpc.unary_unary_rpc_method_handler( |
||||
servicer.getVaslPieces, |
||||
request_deserializer=control__tests__pb2.GetVaslPiecesRequest.FromString, |
||||
response_serializer=control__tests__pb2.GetVaslPiecesResponse.SerializeToString, |
||||
), |
||||
'getAppConfig': grpc.unary_unary_rpc_method_handler( |
||||
servicer.getAppConfig, |
||||
request_deserializer=google_dot_protobuf_dot_empty__pb2.Empty.FromString, |
||||
response_serializer=control__tests__pb2.GetAppConfigResponse.SerializeToString, |
||||
), |
||||
'setAppConfigVal': grpc.unary_unary_rpc_method_handler( |
||||
servicer.setAppConfigVal, |
||||
request_deserializer=control__tests__pb2.SetAppConfigValRequest.FromString, |
||||
response_serializer=google_dot_protobuf_dot_empty__pb2.Empty.SerializeToString, |
||||
), |
||||
'deleteAppConfigVal': grpc.unary_unary_rpc_method_handler( |
||||
servicer.deleteAppConfigVal, |
||||
request_deserializer=control__tests__pb2.DeleteAppConfigValRequest.FromString, |
||||
response_serializer=google_dot_protobuf_dot_empty__pb2.Empty.SerializeToString, |
||||
), |
||||
'saveTempFile': grpc.unary_unary_rpc_method_handler( |
||||
servicer.saveTempFile, |
||||
request_deserializer=control__tests__pb2.SaveTempFileRequest.FromString, |
||||
response_serializer=google_dot_protobuf_dot_empty__pb2.Empty.SerializeToString, |
||||
), |
||||
} |
||||
generic_handler = grpc.method_handlers_generic_handler( |
||||
'ControlTests', rpc_method_handlers) |
||||
server.add_generic_rpc_handlers((generic_handler,)) |
||||
|
||||
|
||||
# This class is part of an EXPERIMENTAL API. |
||||
class ControlTests(object): |
||||
"""-------------------------------------------------------------------- |
||||
|
||||
""" |
||||
|
||||
@staticmethod |
||||
def startTests(request, |
||||
target, |
||||
options=(), |
||||
channel_credentials=None, |
||||
call_credentials=None, |
||||
insecure=False, |
||||
compression=None, |
||||
wait_for_ready=None, |
||||
timeout=None, |
||||
metadata=None): |
||||
return grpc.experimental.unary_unary(request, target, '/ControlTests/startTests', |
||||
google_dot_protobuf_dot_empty__pb2.Empty.SerializeToString, |
||||
control__tests__pb2.StartTestsResponse.FromString, |
||||
options, channel_credentials, |
||||
insecure, call_credentials, compression, wait_for_ready, timeout, metadata) |
||||
|
||||
@staticmethod |
||||
def endTests(request, |
||||
target, |
||||
options=(), |
||||
channel_credentials=None, |
||||
call_credentials=None, |
||||
insecure=False, |
||||
compression=None, |
||||
wait_for_ready=None, |
||||
timeout=None, |
||||
metadata=None): |
||||
return grpc.experimental.unary_unary(request, target, '/ControlTests/endTests', |
||||
google_dot_protobuf_dot_empty__pb2.Empty.SerializeToString, |
||||
google_dot_protobuf_dot_empty__pb2.Empty.FromString, |
||||
options, channel_credentials, |
||||
insecure, call_credentials, compression, wait_for_ready, timeout, metadata) |
||||
|
||||
@staticmethod |
||||
def getVassalVersions(request, |
||||
target, |
||||
options=(), |
||||
channel_credentials=None, |
||||
call_credentials=None, |
||||
insecure=False, |
||||
compression=None, |
||||
wait_for_ready=None, |
||||
timeout=None, |
||||
metadata=None): |
||||
return grpc.experimental.unary_unary(request, target, '/ControlTests/getVassalVersions', |
||||
google_dot_protobuf_dot_empty__pb2.Empty.SerializeToString, |
||||
control__tests__pb2.GetVassalVersionsResponse.FromString, |
||||
options, channel_credentials, |
||||
insecure, call_credentials, compression, wait_for_ready, timeout, metadata) |
||||
|
||||
@staticmethod |
||||
def setVassalVersion(request, |
||||
target, |
||||
options=(), |
||||
channel_credentials=None, |
||||
call_credentials=None, |
||||
insecure=False, |
||||
compression=None, |
||||
wait_for_ready=None, |
||||
timeout=None, |
||||
metadata=None): |
||||
return grpc.experimental.unary_unary(request, target, '/ControlTests/setVassalVersion', |
||||
control__tests__pb2.SetVassalVersionRequest.SerializeToString, |
||||
google_dot_protobuf_dot_empty__pb2.Empty.FromString, |
||||
options, channel_credentials, |
||||
insecure, call_credentials, compression, wait_for_ready, timeout, metadata) |
||||
|
||||
@staticmethod |
||||
def getVaslVersions(request, |
||||
target, |
||||
options=(), |
||||
channel_credentials=None, |
||||
call_credentials=None, |
||||
insecure=False, |
||||
compression=None, |
||||
wait_for_ready=None, |
||||
timeout=None, |
||||
metadata=None): |
||||
return grpc.experimental.unary_unary(request, target, '/ControlTests/getVaslVersions', |
||||
google_dot_protobuf_dot_empty__pb2.Empty.SerializeToString, |
||||
control__tests__pb2.GetVaslVersionsResponse.FromString, |
||||
options, channel_credentials, |
||||
insecure, call_credentials, compression, wait_for_ready, timeout, metadata) |
||||
|
||||
@staticmethod |
||||
def setVaslVersion(request, |
||||
target, |
||||
options=(), |
||||
channel_credentials=None, |
||||
call_credentials=None, |
||||
insecure=False, |
||||
compression=None, |
||||
wait_for_ready=None, |
||||
timeout=None, |
||||
metadata=None): |
||||
return grpc.experimental.unary_unary(request, target, '/ControlTests/setVaslVersion', |
||||
control__tests__pb2.SetVaslVersionRequest.SerializeToString, |
||||
google_dot_protobuf_dot_empty__pb2.Empty.FromString, |
||||
options, channel_credentials, |
||||
insecure, call_credentials, compression, wait_for_ready, timeout, metadata) |
||||
|
||||
@staticmethod |
||||
def getVaslExtns(request, |
||||
target, |
||||
options=(), |
||||
channel_credentials=None, |
||||
call_credentials=None, |
||||
insecure=False, |
||||
compression=None, |
||||
wait_for_ready=None, |
||||
timeout=None, |
||||
metadata=None): |
||||
return grpc.experimental.unary_unary(request, target, '/ControlTests/getVaslExtns', |
||||
google_dot_protobuf_dot_empty__pb2.Empty.SerializeToString, |
||||
control__tests__pb2.GetVaslExtnsResponse.FromString, |
||||
options, channel_credentials, |
||||
insecure, call_credentials, compression, wait_for_ready, timeout, metadata) |
||||
|
||||
@staticmethod |
||||
def setVaslExtnInfoDir(request, |
||||
target, |
||||
options=(), |
||||
channel_credentials=None, |
||||
call_credentials=None, |
||||
insecure=False, |
||||
compression=None, |
||||
wait_for_ready=None, |
||||
timeout=None, |
||||
metadata=None): |
||||
return grpc.experimental.unary_unary(request, target, '/ControlTests/setVaslExtnInfoDir', |
||||
control__tests__pb2.SetVaslExtnInfoDirRequest.SerializeToString, |
||||
google_dot_protobuf_dot_empty__pb2.Empty.FromString, |
||||
options, channel_credentials, |
||||
insecure, call_credentials, compression, wait_for_ready, timeout, metadata) |
||||
|
||||
@staticmethod |
||||
def setGpidRemappings(request, |
||||
target, |
||||
options=(), |
||||
channel_credentials=None, |
||||
call_credentials=None, |
||||
insecure=False, |
||||
compression=None, |
||||
wait_for_ready=None, |
||||
timeout=None, |
||||
metadata=None): |
||||
return grpc.experimental.unary_unary(request, target, '/ControlTests/setGpidRemappings', |
||||
control__tests__pb2.SetGpidRemappingsRequest.SerializeToString, |
||||
google_dot_protobuf_dot_empty__pb2.Empty.FromString, |
||||
options, channel_credentials, |
||||
insecure, call_credentials, compression, wait_for_ready, timeout, metadata) |
||||
|
||||
@staticmethod |
||||
def getVaslModWarnings(request, |
||||
target, |
||||
options=(), |
||||
channel_credentials=None, |
||||
call_credentials=None, |
||||
insecure=False, |
||||
compression=None, |
||||
wait_for_ready=None, |
||||
timeout=None, |
||||
metadata=None): |
||||
return grpc.experimental.unary_unary(request, target, '/ControlTests/getVaslModWarnings', |
||||
google_dot_protobuf_dot_empty__pb2.Empty.SerializeToString, |
||||
control__tests__pb2.GetVaslModWarningsResponse.FromString, |
||||
options, channel_credentials, |
||||
insecure, call_credentials, compression, wait_for_ready, timeout, metadata) |
||||
|
||||
@staticmethod |
||||
def setDataDir(request, |
||||
target, |
||||
options=(), |
||||
channel_credentials=None, |
||||
call_credentials=None, |
||||
insecure=False, |
||||
compression=None, |
||||
wait_for_ready=None, |
||||
timeout=None, |
||||
metadata=None): |
||||
return grpc.experimental.unary_unary(request, target, '/ControlTests/setDataDir', |
||||
control__tests__pb2.SetDataDirRequest.SerializeToString, |
||||
google_dot_protobuf_dot_empty__pb2.Empty.FromString, |
||||
options, channel_credentials, |
||||
insecure, call_credentials, compression, wait_for_ready, timeout, metadata) |
||||
|
||||
@staticmethod |
||||
def setDefaultScenario(request, |
||||
target, |
||||
options=(), |
||||
channel_credentials=None, |
||||
call_credentials=None, |
||||
insecure=False, |
||||
compression=None, |
||||
wait_for_ready=None, |
||||
timeout=None, |
||||
metadata=None): |
||||
return grpc.experimental.unary_unary(request, target, '/ControlTests/setDefaultScenario', |
||||
control__tests__pb2.SetDefaultScenarioRequest.SerializeToString, |
||||
google_dot_protobuf_dot_empty__pb2.Empty.FromString, |
||||
options, channel_credentials, |
||||
insecure, call_credentials, compression, wait_for_ready, timeout, metadata) |
||||
|
||||
@staticmethod |
||||
def setDefaultTemplatePack(request, |
||||
target, |
||||
options=(), |
||||
channel_credentials=None, |
||||
call_credentials=None, |
||||
insecure=False, |
||||
compression=None, |
||||
wait_for_ready=None, |
||||
timeout=None, |
||||
metadata=None): |
||||
return grpc.experimental.unary_unary(request, target, '/ControlTests/setDefaultTemplatePack', |
||||
control__tests__pb2.SetDefaultTemplatePackRequest.SerializeToString, |
||||
google_dot_protobuf_dot_empty__pb2.Empty.FromString, |
||||
options, channel_credentials, |
||||
insecure, call_credentials, compression, wait_for_ready, timeout, metadata) |
||||
|
||||
@staticmethod |
||||
def setVehOrdNotesDir(request, |
||||
target, |
||||
options=(), |
||||
channel_credentials=None, |
||||
call_credentials=None, |
||||
insecure=False, |
||||
compression=None, |
||||
wait_for_ready=None, |
||||
timeout=None, |
||||
metadata=None): |
||||
return grpc.experimental.unary_unary(request, target, '/ControlTests/setVehOrdNotesDir', |
||||
control__tests__pb2.SetVehOrdNotesDirRequest.SerializeToString, |
||||
google_dot_protobuf_dot_empty__pb2.Empty.FromString, |
||||
options, channel_credentials, |
||||
insecure, call_credentials, compression, wait_for_ready, timeout, metadata) |
||||
|
||||
@staticmethod |
||||
def setUserFilesDir(request, |
||||
target, |
||||
options=(), |
||||
channel_credentials=None, |
||||
call_credentials=None, |
||||
insecure=False, |
||||
compression=None, |
||||
wait_for_ready=None, |
||||
timeout=None, |
||||
metadata=None): |
||||
return grpc.experimental.unary_unary(request, target, '/ControlTests/setUserFilesDir', |
||||
control__tests__pb2.SetUserFilesDirRequest.SerializeToString, |
||||
google_dot_protobuf_dot_empty__pb2.Empty.FromString, |
||||
options, channel_credentials, |
||||
insecure, call_credentials, compression, wait_for_ready, timeout, metadata) |
||||
|
||||
@staticmethod |
||||
def setAsaScenarioIndex(request, |
||||
target, |
||||
options=(), |
||||
channel_credentials=None, |
||||
call_credentials=None, |
||||
insecure=False, |
||||
compression=None, |
||||
wait_for_ready=None, |
||||
timeout=None, |
||||
metadata=None): |
||||
return grpc.experimental.unary_unary(request, target, '/ControlTests/setAsaScenarioIndex', |
||||
control__tests__pb2.SetAsaScenarioIndexRequest.SerializeToString, |
||||
google_dot_protobuf_dot_empty__pb2.Empty.FromString, |
||||
options, channel_credentials, |
||||
insecure, call_credentials, compression, wait_for_ready, timeout, metadata) |
||||
|
||||
@staticmethod |
||||
def setRoarScenarioIndex(request, |
||||
target, |
||||
options=(), |
||||
channel_credentials=None, |
||||
call_credentials=None, |
||||
insecure=False, |
||||
compression=None, |
||||
wait_for_ready=None, |
||||
timeout=None, |
||||
metadata=None): |
||||
return grpc.experimental.unary_unary(request, target, '/ControlTests/setRoarScenarioIndex', |
||||
control__tests__pb2.SetRoarScenarioIndexRequest.SerializeToString, |
||||
google_dot_protobuf_dot_empty__pb2.Empty.FromString, |
||||
options, channel_credentials, |
||||
insecure, call_credentials, compression, wait_for_ready, timeout, metadata) |
||||
|
||||
@staticmethod |
||||
def getLastSnippetImage(request, |
||||
target, |
||||
options=(), |
||||
channel_credentials=None, |
||||
call_credentials=None, |
||||
insecure=False, |
||||
compression=None, |
||||
wait_for_ready=None, |
||||
timeout=None, |
||||
metadata=None): |
||||
return grpc.experimental.unary_unary(request, target, '/ControlTests/getLastSnippetImage', |
||||
google_dot_protobuf_dot_empty__pb2.Empty.SerializeToString, |
||||
control__tests__pb2.GetLastSnippetImageResponse.FromString, |
||||
options, channel_credentials, |
||||
insecure, call_credentials, compression, wait_for_ready, timeout, metadata) |
||||
|
||||
@staticmethod |
||||
def resetLastAsaUpload(request, |
||||
target, |
||||
options=(), |
||||
channel_credentials=None, |
||||
call_credentials=None, |
||||
insecure=False, |
||||
compression=None, |
||||
wait_for_ready=None, |
||||
timeout=None, |
||||
metadata=None): |
||||
return grpc.experimental.unary_unary(request, target, '/ControlTests/resetLastAsaUpload', |
||||
google_dot_protobuf_dot_empty__pb2.Empty.SerializeToString, |
||||
google_dot_protobuf_dot_empty__pb2.Empty.FromString, |
||||
options, channel_credentials, |
||||
insecure, call_credentials, compression, wait_for_ready, timeout, metadata) |
||||
|
||||
@staticmethod |
||||
def getLastAsaUpload(request, |
||||
target, |
||||
options=(), |
||||
channel_credentials=None, |
||||
call_credentials=None, |
||||
insecure=False, |
||||
compression=None, |
||||
wait_for_ready=None, |
||||
timeout=None, |
||||
metadata=None): |
||||
return grpc.experimental.unary_unary(request, target, '/ControlTests/getLastAsaUpload', |
||||
google_dot_protobuf_dot_empty__pb2.Empty.SerializeToString, |
||||
control__tests__pb2.GetLastAsaUploadResponse.FromString, |
||||
options, channel_credentials, |
||||
insecure, call_credentials, compression, wait_for_ready, timeout, metadata) |
||||
|
||||
@staticmethod |
||||
def dumpVsav(request, |
||||
target, |
||||
options=(), |
||||
channel_credentials=None, |
||||
call_credentials=None, |
||||
insecure=False, |
||||
compression=None, |
||||
wait_for_ready=None, |
||||
timeout=None, |
||||
metadata=None): |
||||
return grpc.experimental.unary_unary(request, target, '/ControlTests/dumpVsav', |
||||
control__tests__pb2.DumpVsavRequest.SerializeToString, |
||||
control__tests__pb2.DumpVsavResponse.FromString, |
||||
options, channel_credentials, |
||||
insecure, call_credentials, compression, wait_for_ready, timeout, metadata) |
||||
|
||||
@staticmethod |
||||
def getVaslPieces(request, |
||||
target, |
||||
options=(), |
||||
channel_credentials=None, |
||||
call_credentials=None, |
||||
insecure=False, |
||||
compression=None, |
||||
wait_for_ready=None, |
||||
timeout=None, |
||||
metadata=None): |
||||
return grpc.experimental.unary_unary(request, target, '/ControlTests/getVaslPieces', |
||||
control__tests__pb2.GetVaslPiecesRequest.SerializeToString, |
||||
control__tests__pb2.GetVaslPiecesResponse.FromString, |
||||
options, channel_credentials, |
||||
insecure, call_credentials, compression, wait_for_ready, timeout, metadata) |
||||
|
||||
@staticmethod |
||||
def getAppConfig(request, |
||||
target, |
||||
options=(), |
||||
channel_credentials=None, |
||||
call_credentials=None, |
||||
insecure=False, |
||||
compression=None, |
||||
wait_for_ready=None, |
||||
timeout=None, |
||||
metadata=None): |
||||
return grpc.experimental.unary_unary(request, target, '/ControlTests/getAppConfig', |
||||
google_dot_protobuf_dot_empty__pb2.Empty.SerializeToString, |
||||
control__tests__pb2.GetAppConfigResponse.FromString, |
||||
options, channel_credentials, |
||||
insecure, call_credentials, compression, wait_for_ready, timeout, metadata) |
||||
|
||||
@staticmethod |
||||
def setAppConfigVal(request, |
||||
target, |
||||
options=(), |
||||
channel_credentials=None, |
||||
call_credentials=None, |
||||
insecure=False, |
||||
compression=None, |
||||
wait_for_ready=None, |
||||
timeout=None, |
||||
metadata=None): |
||||
return grpc.experimental.unary_unary(request, target, '/ControlTests/setAppConfigVal', |
||||
control__tests__pb2.SetAppConfigValRequest.SerializeToString, |
||||
google_dot_protobuf_dot_empty__pb2.Empty.FromString, |
||||
options, channel_credentials, |
||||
insecure, call_credentials, compression, wait_for_ready, timeout, metadata) |
||||
|
||||
@staticmethod |
||||
def deleteAppConfigVal(request, |
||||
target, |
||||
options=(), |
||||
channel_credentials=None, |
||||
call_credentials=None, |
||||
insecure=False, |
||||
compression=None, |
||||
wait_for_ready=None, |
||||
timeout=None, |
||||
metadata=None): |
||||
return grpc.experimental.unary_unary(request, target, '/ControlTests/deleteAppConfigVal', |
||||
control__tests__pb2.DeleteAppConfigValRequest.SerializeToString, |
||||
google_dot_protobuf_dot_empty__pb2.Empty.FromString, |
||||
options, channel_credentials, |
||||
insecure, call_credentials, compression, wait_for_ready, timeout, metadata) |
||||
|
||||
@staticmethod |
||||
def saveTempFile(request, |
||||
target, |
||||
options=(), |
||||
channel_credentials=None, |
||||
call_credentials=None, |
||||
insecure=False, |
||||
compression=None, |
||||
wait_for_ready=None, |
||||
timeout=None, |
||||
metadata=None): |
||||
return grpc.experimental.unary_unary(request, target, '/ControlTests/saveTempFile', |
||||
control__tests__pb2.SaveTempFileRequest.SerializeToString, |
||||
google_dot_protobuf_dot_empty__pb2.Empty.FromString, |
||||
options, channel_credentials, |
||||
insecure, call_credentials, compression, wait_for_ready, timeout, metadata) |
||||
@ -0,0 +1,7 @@ |
||||
""" Injected functions for GetVaslPiecesRequest. """ |
||||
|
||||
# --------------------------------------------------------------------- |
||||
|
||||
def brief( self ): |
||||
"""Return a GetVaslPiecesRequest as a brief string.""" |
||||
return self.vaslVersion |
||||
@ -0,0 +1,7 @@ |
||||
""" Injected functions for SaveTempFileRequest. """ |
||||
|
||||
# --------------------------------------------------------------------- |
||||
|
||||
def brief( self ): |
||||
"""Return a SaveTempFileRequest as a brief string.""" |
||||
return "{} (#bytes={})".format( self.fileName, len(self.data) ) |
||||
@ -0,0 +1,14 @@ |
||||
""" Injected functions for SetAppConfigValRequest. """ |
||||
|
||||
# --------------------------------------------------------------------- |
||||
|
||||
def brief( self ): |
||||
"""Return a SetAppConfigValRequest as a brief string.""" |
||||
for val_type in ( "strVal", "intVal", "boolVal" ): |
||||
if self.HasField( val_type ): |
||||
val = getattr( self, val_type ) |
||||
return "{} = {} ({})".format( |
||||
self.key, val, type(val).__name__ |
||||
) |
||||
assert False |
||||
return "{} = ???".format( self.key ) |
||||
@ -0,0 +1,7 @@ |
||||
""" Injected functions for SetAsaScenarioIndexRequest. """ |
||||
|
||||
# --------------------------------------------------------------------- |
||||
|
||||
def brief( self ): |
||||
"""Return a SetAsaScenarioIndexRequest as a brief string.""" |
||||
return self.fileName |
||||
@ -0,0 +1,10 @@ |
||||
""" Injected functions for SetDataDirRequest. """ |
||||
|
||||
from .generated.control_tests_pb2 import SetDataDirRequest |
||||
from .utils import enum_to_string |
||||
|
||||
# --------------------------------------------------------------------- |
||||
|
||||
def brief( self ): |
||||
"""Return a SetDataDirRequest as a brief string.""" |
||||
return enum_to_string( SetDataDirRequest.DirType, self.dirType ) #pylint: disable=no-member |
||||
@ -0,0 +1,7 @@ |
||||
""" Injected functions for SetDefaultScenarioRequest. """ |
||||
|
||||
# --------------------------------------------------------------------- |
||||
|
||||
def brief( self ): |
||||
"""Return a SetDefaultScenarioRequest as a brief string.""" |
||||
return self.fileName |
||||
@ -0,0 +1,21 @@ |
||||
""" Injected functions for SetDefaultTemplatePackRequest. """ |
||||
|
||||
from vasl_templates.webapp.tests.proto.generated.control_tests_pb2 import SetDefaultTemplatePackRequest |
||||
from .utils import enum_to_string |
||||
|
||||
# --------------------------------------------------------------------- |
||||
|
||||
def brief( self ): |
||||
"""Return a SetDefaultTemplatePackRequest as a brief string.""" |
||||
if self.HasField( "templatePackType" ): |
||||
return enum_to_string( |
||||
SetDefaultTemplatePackRequest.TemplatePackType, #pylint: disable=no-member |
||||
self.templatePackType |
||||
) |
||||
elif self.HasField( "dirName" ): |
||||
return self.dirName |
||||
elif self.HasField( "zipData" ): |
||||
return "zip: #bytes={}".format( len(self.zipData) ) |
||||
else: |
||||
assert False |
||||
return str( self ).strip() |
||||
@ -0,0 +1,7 @@ |
||||
""" Injected functions for SetRoarScenarioIndexRequest. """ |
||||
|
||||
# --------------------------------------------------------------------- |
||||
|
||||
def brief( self ): |
||||
"""Return a SetRoarScenarioIndexRequest as a brief string.""" |
||||
return self.fileName |
||||
@ -0,0 +1,7 @@ |
||||
""" Injected functions for SetUserFilesDirRequest. """ |
||||
|
||||
# --------------------------------------------------------------------- |
||||
|
||||
def brief( self ): |
||||
"""Return a SetUserFilesDirRequest as a brief string.""" |
||||
return self.dirOrUrl |
||||
@ -0,0 +1,7 @@ |
||||
""" Injected functions for SetVaslExtnInfoDirRequest. """ |
||||
|
||||
# --------------------------------------------------------------------- |
||||
|
||||
def brief( self ): |
||||
"""Return a SetVaslExtnInfoDirRequest as a brief string.""" |
||||
return self.dirName |
||||
@ -0,0 +1,13 @@ |
||||
""" Injected functions for SetVaslVersionRequest. """ |
||||
|
||||
from vasl_templates.webapp.tests.proto.generated.control_tests_pb2 import SetVaslVersionRequest |
||||
from .utils import enum_to_string |
||||
|
||||
# --------------------------------------------------------------------- |
||||
|
||||
def brief( self ): |
||||
"""Return a SetVaslVersionRequest as a brief string.""" |
||||
return "{} (extns={})".format( |
||||
self.vaslVersion, |
||||
enum_to_string( SetVaslVersionRequest.VaslExtnsType, self.vaslExtnsType ) #pylint: disable=no-member |
||||
) |
||||
@ -0,0 +1,7 @@ |
||||
""" Injected functions for SetVassalVersionRequest. """ |
||||
|
||||
# --------------------------------------------------------------------- |
||||
|
||||
def brief( self ): |
||||
"""Return a SetVassalVersionRequest as a brief string.""" |
||||
return self.vassalVersion |
||||
@ -0,0 +1,10 @@ |
||||
""" Injected functions for SetVehOrdNotesDirRequest. """ |
||||
|
||||
from vasl_templates.webapp.tests.proto.generated.control_tests_pb2 import SetVehOrdNotesDirRequest |
||||
from .utils import enum_to_string |
||||
|
||||
# --------------------------------------------------------------------- |
||||
|
||||
def brief( self ): |
||||
"""Return a SetVehOrdNotesDirRequest as a brief string.""" |
||||
return enum_to_string( SetVehOrdNotesDirRequest.DirType, self.dirType ) #pylint: disable=no-member |
||||
@ -0,0 +1,44 @@ |
||||
"""Utility functions.""" |
||||
|
||||
import inspect |
||||
|
||||
# --------------------------------------------------------------------- |
||||
|
||||
def get_classes(): |
||||
"""Get the request/response classes.""" |
||||
from .generated import control_tests_pb2 |
||||
for elem in dir( control_tests_pb2 ): |
||||
if not inspect.isclass( type(elem) ): |
||||
continue |
||||
if not elem.endswith( ( "Request", "Response" ) ): |
||||
continue |
||||
cls = getattr( control_tests_pb2, elem ) |
||||
yield cls |
||||
|
||||
# --------------------------------------------------------------------- |
||||
|
||||
def split_words( val ): |
||||
"""Extract words from a camel-cased string.""" |
||||
words, curr_word = [], [] |
||||
for ch in val: |
||||
if ch.isupper(): |
||||
if curr_word: |
||||
words.append( "".join( curr_word ) ) |
||||
curr_word = [] |
||||
curr_word.append( ch.lower() ) |
||||
if curr_word: |
||||
words.append( "".join( curr_word ) ) |
||||
return words |
||||
|
||||
# --------------------------------------------------------------------- |
||||
|
||||
def enum_to_string( enum, val ): |
||||
"""Convert an enum value to a string.""" |
||||
val = enum.Name( val ) |
||||
return "{%s}" % val |
||||
|
||||
def enum_from_string( enum, val ): |
||||
"""Convert a string to an enum value.""" |
||||
if not val.startswith( "{" ) or not val.endswith( "}" ): |
||||
raise ValueError( "Invalid enumerated value for {}: {}".format( enum.DESCRIPTOR.full_name, val ) ) |
||||
return enum.Value( val[1:-1] ) |
||||
@ -1,373 +0,0 @@ |
||||
"""Allow a remote server to be controlled during tests. |
||||
|
||||
We sometimes make changes to the webapp server during tests, and while we used to do that using pytest's monkeypatch, |
||||
that will not work if we are talking to a remote (i.e. in another process) server. This module defines the things |
||||
that can be changed during the course of tests, and a simple RPC mechanism that lets them be executed remotely. |
||||
It needs to be enabled in the server via the ENABLE_REMOTE_TEST_CONTROL debug switch. |
||||
""" |
||||
|
||||
import os |
||||
import urllib.request |
||||
import json |
||||
import glob |
||||
import base64 |
||||
import tempfile |
||||
import logging |
||||
import random |
||||
|
||||
import pytest |
||||
|
||||
from vasl_templates.webapp import app, globvars |
||||
from vasl_templates.webapp.config.constants import DATA_DIR |
||||
from vasl_templates.webapp.vassal import VassalShim |
||||
from vasl_templates.webapp.vasl_mod import set_vasl_mod |
||||
from vasl_templates.webapp.utils import TempFile |
||||
from vasl_templates.webapp import main as webapp_main |
||||
from vasl_templates.webapp import snippets as webapp_snippets |
||||
from vasl_templates.webapp import scenarios as webapp_scenarios |
||||
from vasl_templates.webapp import vasl_mod as vasl_mod_module |
||||
from vasl_templates.webapp import vo_utils as vo_utils_module |
||||
|
||||
_logger = logging.getLogger( "control_tests" ) |
||||
|
||||
_ORIG_CHAPTER_H_NOTES_DIR = app.config.get( "CHAPTER_H_NOTES_DIR", os.environ.get("CHAPTER_H_NOTES_DIR") ) |
||||
|
||||
# --------------------------------------------------------------------- |
||||
|
||||
class ControlTests: |
||||
"""Control a remote server during tests.""" |
||||
|
||||
def __init__( self, webapp ): |
||||
self.webapp = webapp |
||||
try: |
||||
self.server_url = pytest.config.option.server_url #pylint: disable=no-member |
||||
except AttributeError: |
||||
self.server_url = None |
||||
# set up a directory for our temp files |
||||
self._temp_dir = tempfile.TemporaryDirectory() |
||||
|
||||
def __del__( self ): |
||||
self._temp_dir.cleanup() |
||||
|
||||
def __getattr__( self, name ): |
||||
"""Generic entry point for handling control requests.""" |
||||
if name.startswith( ("get_","set_","reset_") ): |
||||
# check if we are talking to a local or remote server |
||||
if self.server_url: |
||||
# remote: return a function that will invoke the handler function on the remote server |
||||
def call_remote( **kwargs ): #pylint: disable=missing-docstring |
||||
return self._remote_test_control( name, **kwargs ) |
||||
return call_remote |
||||
else: |
||||
# local: return the actual handler function |
||||
return getattr( self, "_"+name ) |
||||
raise AttributeError( name ) |
||||
|
||||
def _remote_test_control( self, action, **kwargs ): |
||||
"""Invoke a handler function on the remote server.""" |
||||
if "bin_data" in kwargs: |
||||
kwargs["bin_data"] = base64.b64encode( kwargs["bin_data"] ) |
||||
if "gpids" in kwargs: |
||||
kwargs["gpids"] = json.dumps( kwargs["gpids"] ) |
||||
resp = urllib.request.urlopen( |
||||
self.webapp.url_for( "control_tests", action=action, **kwargs ) |
||||
).read() |
||||
if resp == b"ok": |
||||
return self |
||||
else: |
||||
return json.loads( resp.decode( "utf-8" ) ) |
||||
|
||||
def _get_app_config( self ): #pylint: disable=no-self-use |
||||
"""Get the webapp config.""" |
||||
return { |
||||
k: v for k,v in app.config.items() |
||||
if isinstance( v, (str,int,bool,list,dict) ) |
||||
} |
||||
|
||||
def _set_app_config( self, key=None, val=None ): |
||||
"""Set the webapp config.""" |
||||
if val is None: |
||||
del app.config[ key ] |
||||
else: |
||||
app.config[ key ] = val |
||||
return self |
||||
|
||||
def _set_data_dir( self, dtype=None ): |
||||
"""Set the webapp's data directory.""" |
||||
if dtype == "real": |
||||
dname = DATA_DIR |
||||
elif dtype == "test": |
||||
dname = os.path.join( os.path.split(__file__)[0], "fixtures/data" ) |
||||
else: |
||||
raise RuntimeError( "Unknown data dir type: {}".format( dtype ) ) |
||||
_logger.info( "Setting data dir: %s", dname ) |
||||
self.webapp.config[ "DATA_DIR" ] = dname |
||||
vo_utils_module._vo_comments = None #pylint: disable=protected-access |
||||
from vasl_templates.webapp.vo import load_vo_listings |
||||
load_vo_listings( None ) |
||||
return self |
||||
|
||||
def _set_default_scenario( self, fname=None ): |
||||
"""Set the default scenario.""" |
||||
if fname: |
||||
dname = os.path.join( os.path.split(__file__)[0], "fixtures" ) |
||||
fname = os.path.join( dname, fname ) |
||||
_logger.info( "Setting default scenario: %s", fname ) |
||||
webapp_main.default_scenario = fname |
||||
return self |
||||
|
||||
def _set_default_template_pack( self, dname=None, bin_data=None ): |
||||
"""Set the default template pack.""" |
||||
if bin_data: |
||||
fname = os.path.join( self._temp_dir.name, "default-template-pack.zip" ) |
||||
with open( fname, "wb" ) as fp: |
||||
fp.write( bin_data ) |
||||
dname = fname |
||||
elif dname == "real": |
||||
dname = os.path.join( os.path.split(__file__)[0], "../data/default-template-pack" ) |
||||
elif dname: |
||||
dname2 = os.path.join( os.path.split(__file__)[0], "fixtures" ) |
||||
dname = os.path.join( dname2, dname ) |
||||
_logger.info( "Setting default template pack: %s", dname ) |
||||
webapp_snippets.default_template_pack = dname |
||||
return self |
||||
|
||||
def _set_gpid_remappings( self, gpids=None ): #pylint: disable=no-self-use |
||||
"""Configure the GPID remappings.""" |
||||
if isinstance( gpids, str ): |
||||
gpids = json.loads( gpids.replace( "'", '"' ) ) |
||||
for row in gpids: |
||||
row[1] = { str(k): v for k,v in row[1].items() } |
||||
_logger.info( "Setting GPID remappings: %s", gpids ) |
||||
prev_gpid_mappings = vasl_mod_module.GPID_REMAPPINGS |
||||
vasl_mod_module.GPID_REMAPPINGS = gpids |
||||
return prev_gpid_mappings |
||||
|
||||
def _get_vasl_mods( self ): |
||||
"""Return the available VASL modules.""" |
||||
fnames = self._do_get_vasl_mods() |
||||
_logger.debug( "Returning VASL modules:\n%s", |
||||
"\n".join( "- {}".format( f ) for f in fnames ) |
||||
) |
||||
return fnames |
||||
|
||||
def _do_get_vasl_mods( self ): #pylint: disable=no-self-use |
||||
"""Return the available VASL modules.""" |
||||
try: |
||||
dname = pytest.config.option.vasl_mods #pylint: disable=no-member |
||||
assert dname, "--vasl-mods was not specified." |
||||
except AttributeError: |
||||
dname = app.config[ "TEST_VASL_MODS" ] |
||||
fspec = os.path.join( dname, "*.vmod" ) |
||||
return glob.glob( fspec ) |
||||
|
||||
def _set_vasl_mod( self, vmod=None, extns_dtype=None ): |
||||
"""Install a VASL module.""" |
||||
|
||||
# configure the VASL extensions |
||||
if extns_dtype: |
||||
if extns_dtype == "real": |
||||
try: |
||||
dname = pytest.config.option.vasl_extensions #pylint: disable=no-member |
||||
assert dname, "--vasl-extensions was not specified." |
||||
except AttributeError: |
||||
dname = app.config[ "TEST_VASL_EXTNS_DIR" ] |
||||
elif extns_dtype == "test": |
||||
dname = self._temp_dir.name |
||||
else: |
||||
assert False, "Unknown extensions directory type: "+extns_dtype |
||||
_logger.info( "Enabling VASL extensions: %s", dname ) |
||||
app.config[ "VASL_EXTNS_DIR" ] = dname |
||||
else: |
||||
_logger.info( "Disabling VASL extensions." ) |
||||
app.config[ "VASL_EXTNS_DIR" ] = None |
||||
|
||||
# configure the VASL module |
||||
if vmod: |
||||
vmod_fnames = self._do_get_vasl_mods() |
||||
if vmod == "random": |
||||
# NOTE: Some tests require a VASL module to be loaded, and since they should all |
||||
# should behave in the same way, it doesn't matter which one we load. |
||||
vmod = random.choice( vmod_fnames ) |
||||
else: |
||||
assert vmod in vmod_fnames |
||||
app.config[ "VASL_MOD" ] = vmod |
||||
else: |
||||
app.config[ "VASL_MOD" ] = None |
||||
_logger.info( "Installing VASL module: %s", vmod ) |
||||
|
||||
# install the new VASL module |
||||
from vasl_templates.webapp.main import startup_msg_store |
||||
startup_msg_store.reset() |
||||
vasl_mod_module.warnings = [] |
||||
set_vasl_mod( vmod, startup_msg_store ) |
||||
from vasl_templates.webapp.vo import load_vo_listings |
||||
load_vo_listings( None ) |
||||
|
||||
return self |
||||
|
||||
def _get_vasl_extns( self ): #pylint: disable=no-self-use |
||||
"""Return the loaded VASL extensions.""" |
||||
extns = globvars.vasl_mod.get_extns() |
||||
_logger.debug( "Returning VASL extensions:\n%s", |
||||
"\n".join( "- {}".format( e ) for e in extns ) |
||||
) |
||||
return extns |
||||
|
||||
def _set_test_vasl_extn( self, fname=None, bin_data=None ): |
||||
"""Set the test VASL extension.""" |
||||
fname = os.path.join( self._temp_dir.name, fname ) |
||||
with open( fname, "wb" ) as fp: |
||||
fp.write( bin_data ) |
||||
return self |
||||
|
||||
def _set_vasl_extn_info_dir( self, dtype=None ): |
||||
"""Set the directory containing the VASL extension info files.""" |
||||
if dtype: |
||||
dname = os.path.join( os.path.split(__file__)[0], "fixtures/vasl-extensions" ) |
||||
dname = os.path.join( dname, dtype ) |
||||
_logger.info( "Setting the default VASL extension info directory: %s", dname ) |
||||
app.config[ "_VASL_EXTN_INFO_DIR_" ] = dname |
||||
else: |
||||
_logger.info( "Using the default VASL extension info directory." ) |
||||
app.config[ "_VASL_EXTN_INFO_DIR_" ] = None |
||||
return self |
||||
|
||||
def _get_vassal_engines( self ): |
||||
"""Get the available VASSAL engines.""" |
||||
vassal_engines = self._do_get_vassal_engines() |
||||
_logger.debug( "Returning VASSAL engines:\n%s", |
||||
"\n".join( "- {}".format( ve ) for ve in vassal_engines ) |
||||
) |
||||
return vassal_engines |
||||
|
||||
def _do_get_vassal_engines( self ): #pylint: disable=no-self-use |
||||
"""Get the available VASSAL engines.""" |
||||
try: |
||||
dname = pytest.config.option.vassal #pylint: disable=no-member |
||||
assert dname, "--vassal was not specified." |
||||
except AttributeError: |
||||
dname = app.config[ "TEST_VASSAL_ENGINES" ] |
||||
vassal_engines = [] |
||||
for root,_,fnames in os.walk( dname ): |
||||
if os.sep+"_disabled_"+os.sep in root: |
||||
continue |
||||
for fname in fnames: |
||||
if fname == "Vengine.jar": |
||||
if root.endswith( "/lib" ): |
||||
root = root[:-4] |
||||
vassal_engines.append( root ) |
||||
return vassal_engines |
||||
|
||||
def _set_vassal_engine( self, vengine=None ): |
||||
"""Install a VASSAL engine.""" |
||||
if vengine: |
||||
assert vengine in self._do_get_vassal_engines() |
||||
_logger.info( "Installing VASSAL engine: %s", vengine ) |
||||
app.config["VASSAL_DIR"] = vengine |
||||
return self |
||||
|
||||
def _get_vsav_dump( self, bin_data=None ): #pylint: disable=no-self-use |
||||
"""Dump a VSAV file.""" |
||||
with TempFile( mode="wb" ) as temp_file: |
||||
temp_file.write( bin_data ) |
||||
temp_file.close( delete=False ) |
||||
vassal_shim = VassalShim() |
||||
return vassal_shim.dump_scenario( temp_file.name ) |
||||
|
||||
def _set_vo_notes_dir( self, dtype=None ): |
||||
"""Set the vehicle/ordnance notes directory.""" |
||||
if dtype == "real": |
||||
try: |
||||
dname = pytest.config.option.vo_notes #pylint: disable=no-member |
||||
assert dname, "--vo-notes was not specified." |
||||
except AttributeError: |
||||
dname = _ORIG_CHAPTER_H_NOTES_DIR |
||||
elif dtype == "test": |
||||
dname = os.path.join( os.path.split(__file__)[0], "fixtures/vo-notes" ) |
||||
else: |
||||
assert dtype is None |
||||
dname = None |
||||
_logger.info( "Setting vehicle/ordnance notes: %s", dname ) |
||||
app.config["CHAPTER_H_NOTES_DIR"] = dname |
||||
from vasl_templates.webapp.vo_notes import load_vo_notes |
||||
load_vo_notes( None ) |
||||
return self |
||||
|
||||
def _set_user_files_dir( self, dtype=None ): |
||||
"""Set the user files directory.""" |
||||
if dtype == "test": |
||||
dname = os.path.join( os.path.split(__file__)[0], "fixtures/user-files" ) |
||||
elif dtype and dtype.startswith( ("http://","https://") ): |
||||
dname = dtype |
||||
else: |
||||
assert dtype is None |
||||
dname = None |
||||
_logger.info( "Setting user files: %s", dname ) |
||||
app.config["USER_FILES_DIR"] = dname |
||||
return self |
||||
|
||||
def _get_last_snippet_image( self ): #pylint: disable=no-self-use |
||||
"""Get the last snippet image generated.""" |
||||
from vasl_templates.webapp.snippets import last_snippet_image |
||||
assert last_snippet_image |
||||
_logger.info( "Returning the last snippet image: #bytes=%d", len(last_snippet_image) ) |
||||
return base64.b64encode( last_snippet_image ).decode( "utf-8" ) |
||||
|
||||
def _get_vasl_mod_warnings( self ): #pylint: disable=no-self-use |
||||
"""Get the vasl_mod startup warnings.""" |
||||
_logger.info( "Returning the vasl_mod startup warnings: %s", vasl_mod_module.warnings ) |
||||
return vasl_mod_module.warnings |
||||
|
||||
def _reset_template_pack( self ): |
||||
"""Force the default template pack to be reloaded.""" |
||||
_logger.info( "Reseting the default template pack." ) |
||||
globvars.template_pack = None |
||||
return self |
||||
|
||||
def _set_roar_scenario_index( self, fname=None ): |
||||
"""Set the ROAR scenario index file.""" |
||||
if fname: |
||||
dname = os.path.join( os.path.split(__file__)[0], "fixtures" ) |
||||
fname = os.path.join( dname, fname ) |
||||
_logger.info( "Setting the ROAR scenario index file: %s", fname ) |
||||
webapp_scenarios._roar_scenarios._set_data( fname ) #pylint: disable=protected-access |
||||
else: |
||||
assert False |
||||
return self |
||||
|
||||
def _set_scenario_index( self, fname=None ): |
||||
"""Set the scenario index file.""" |
||||
if fname: |
||||
dname = os.path.join( os.path.split(__file__)[0], "fixtures" ) |
||||
fname = os.path.join( dname, fname ) |
||||
_logger.info( "Setting the scenario index file: %s", fname ) |
||||
webapp_scenarios._asa_scenarios._set_data( fname ) #pylint: disable=protected-access |
||||
else: |
||||
assert False |
||||
return self |
||||
|
||||
def _reset_last_asa_upload( self ): |
||||
"""Reset the saved last upload to the ASL Scenario Archive.""" |
||||
_logger.info( "Reseting the last ASA upload." ) |
||||
webapp_scenarios._last_asa_upload = None #pylint: disable=protected-access |
||||
return self |
||||
|
||||
def _get_last_asa_upload( self ): #pylint: disable=no-self-use |
||||
"""Get the last set of files uploaded to the ASL Scenario Archive.""" |
||||
last_asa_upload = webapp_scenarios._last_asa_upload #pylint: disable=protected-access |
||||
if not last_asa_upload: |
||||
return {} # FUDGE! This is for the remote testing framework :-/ |
||||
last_asa_upload = last_asa_upload.copy() |
||||
# FUDGE! We can't send binary data over the remote testing interface, but since the tests just check |
||||
# for the presence of a VASL save file and screenshot, we just send an indicator of that, and not |
||||
# the data itself. This will be reworked when we switch to using gRPC. |
||||
if "vasl_setup" in last_asa_upload: |
||||
assert last_asa_upload["vasl_setup"][:2] == b"PK" |
||||
last_asa_upload[ "vasl_setup" ] = "PK:{}".format( len(last_asa_upload["vasl_setup"]) ) |
||||
if "screenshot" in last_asa_upload: |
||||
assert last_asa_upload["screenshot"][:2] == b"\xff\xd8" \ |
||||
and last_asa_upload["screenshot"][-2:] == b"\xff\xd9" # nb: these are the magic numbers for JPEG's |
||||
last_asa_upload[ "screenshot" ] = "JPEG:{}".format( len(last_asa_upload["screenshot"]) ) |
||||
_logger.debug( "Returning the last ASA upload: %s", last_asa_upload ) |
||||
return last_asa_upload |
||||
Loading…
Reference in new issue