forked from 5genesis/ELCM
-
Notifications
You must be signed in to change notification settings - Fork 1
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge pull request #4 from EVOLVED-5G/testcases_v2
Testcases v2
- Loading branch information
Showing
9 changed files
with
408 additions
and
178 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,5 @@ | ||
from .loader_base import Loader | ||
from .resource_loader import ResourceLoader | ||
from .scenario_loader import ScenarioLoader | ||
from .ue_loader import UeLoader | ||
from .testcase_loader import TestCaseLoader |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,77 @@ | ||
import yaml | ||
from Helper import Level | ||
from typing import List, Dict | ||
from os.path import join | ||
from ..action_information import ActionInformation | ||
|
||
|
||
class Loader: | ||
@classmethod | ||
def EnsureFolder(cls, path: str) -> [(Level, str)]: | ||
from Helper import IO | ||
validation = [] | ||
if not IO.EnsureFolder(path): | ||
validation.append((Level.INFO, f'Auto-generated folder: {path}')) | ||
return validation | ||
|
||
@classmethod | ||
def LoadFolder(cls, path: str, kind: str) -> [(Level, str)]: | ||
from Helper import IO | ||
ignored = [] | ||
validation = [] | ||
for file in IO.ListFiles(path): | ||
if file.endswith('.yml'): | ||
filePath = join(path, file) | ||
try: | ||
validation.append((Level.INFO, f'Loading {kind}: {file}')) | ||
data, v = cls.LoadFile(filePath) | ||
validation.extend(v) | ||
validation.extend(cls.ProcessData(data)) | ||
except Exception as e: | ||
validation.append((Level.ERROR, f"Exception loading {kind} file '{filePath}': {e}")) | ||
else: | ||
ignored.append(file) | ||
if len(ignored) != 0: | ||
validation.append((Level.WARNING, | ||
f'Ignored the following files on the {kind}s folder: {(", ".join(ignored))}')) | ||
return validation | ||
|
||
@classmethod | ||
def LoadFile(cls, path: str) -> ((Dict | None), [(Level, str)]): | ||
try: | ||
with open(path, 'r', encoding='utf-8') as file: | ||
raw = yaml.safe_load(file) | ||
return raw, [] | ||
except Exception as e: | ||
return None, [(Level.ERROR, f"Unable to load file '{path}': {e}")] | ||
|
||
@classmethod | ||
def GetActionList(cls, data: List[Dict]) -> ([ActionInformation], [(Level, str)]): | ||
actionList = [] | ||
validation = [] | ||
|
||
for action in data: | ||
actionInfo = ActionInformation.FromMapping(action) | ||
if actionInfo is not None: | ||
actionList.append(actionInfo) | ||
else: | ||
validation.append((Level.ERROR, f'Action not correctly defined for element (data="{action}").')) | ||
actionList.append(ActionInformation.MessageAction( | ||
'ERROR', f'Incorrect Action (data="{action}")' | ||
)) | ||
|
||
if len(actionList) == 0: | ||
validation.append((Level.WARNING, 'No actions defined')) | ||
else: | ||
for action in actionList: | ||
validation.append((Level.DEBUG, str(action))) | ||
|
||
return actionList, validation | ||
|
||
@classmethod | ||
def ProcessData(cls, data: Dict) -> [(Level, str)]: | ||
raise NotImplementedError | ||
|
||
@classmethod | ||
def Clear(cls): | ||
raise NotImplementedError |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,27 @@ | ||
from Helper import Level | ||
from .loader_base import Loader | ||
from ..resource import Resource | ||
from typing import Dict | ||
|
||
|
||
class ResourceLoader(Loader): | ||
resources: Dict[str, Resource] = {} | ||
|
||
@classmethod | ||
def ProcessData(cls, data: Dict) -> [(Level, str)]: | ||
validation = [] | ||
|
||
resource = Resource(data) | ||
if resource.Id in cls.resources.keys(): | ||
validation.append((Level.WARNING, f'Redefining Resource {resource.Id}')) | ||
cls.resources[resource.Id] = resource | ||
|
||
return validation | ||
|
||
@classmethod | ||
def Clear(cls): | ||
cls.resources = {} | ||
|
||
@classmethod | ||
def GetCurrentResources(cls): | ||
return cls.resources |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,31 @@ | ||
from Helper import Level | ||
from .loader_base import Loader | ||
from typing import Dict | ||
|
||
|
||
class ScenarioLoader(Loader): | ||
scenarios: Dict[str, Dict] = {} | ||
|
||
@classmethod | ||
def ProcessData(cls, data: Dict) -> [(Level, str)]: | ||
validation = [] | ||
keys = list(data.keys()) | ||
|
||
if len(keys) > 1: | ||
validation.append((Level.WARNING, f'Multiple Scenarios defined on a single file: {keys}')) | ||
|
||
for key, value in data.items(): | ||
if key in cls.scenarios.keys(): | ||
validation.append((Level.WARNING, f'Redefining Scenario {key}')) | ||
cls.scenarios[key] = value | ||
validation.append((Level.DEBUG, f'{key}: {value}')) | ||
|
||
return validation | ||
|
||
@classmethod | ||
def Clear(cls): | ||
cls.scenarios = {} | ||
|
||
@classmethod | ||
def GetCurrentScenarios(cls): | ||
return cls.scenarios |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,165 @@ | ||
from Helper import Level | ||
from .loader_base import Loader | ||
from ..action_information import ActionInformation | ||
from ..dashboard_panel import DashboardPanel | ||
from typing import Dict, List, Tuple | ||
|
||
|
||
class TestCaseData: | ||
def __init__(self, data: Dict): | ||
# Shared keys | ||
self.AllKeys: List[str] = list(data.keys()) | ||
self.Dashboard: (Dict | None) = data.pop('Dashboard', None) | ||
self.Standard: (bool | None) = data.pop('Standard', None) | ||
self.Custom: (List[str] | None) = data.pop('Custom', None) | ||
self.Distributed: bool = data.pop('Distributed', False) | ||
self.Parameters: Dict[str, Dict[str, str]] = data.pop('Parameters', {}) | ||
|
||
# V2 only | ||
self.Name: (str | None) = data.pop('Name', None) | ||
self.Sequence: List[Dict] = data.pop('Sequence', []) | ||
|
||
|
||
class TestCaseLoader(Loader): | ||
testCases: Dict[str, List[ActionInformation]] = {} | ||
extra: Dict[str, Dict[str, object]] = {} | ||
dashboards: Dict[str, List[DashboardPanel]] = {} | ||
parameters: Dict[str, Tuple[str, str]] = {} # For use only while processing data, not necessary afterwards | ||
|
||
@classmethod | ||
def getPanelList(cls, data: Dict) -> ([DashboardPanel], [(Level, str)]): | ||
validation = [] | ||
panelList = [] | ||
for panel in data: | ||
try: | ||
parsedPanel = DashboardPanel(panel) | ||
valid, error = parsedPanel.Validate() | ||
if not valid: | ||
validation.append((Level.ERROR, f'Could not validate panel (data={panel}) - {error}')) | ||
else: | ||
panelList.append(parsedPanel) | ||
except Exception as e: | ||
validation.append((Level.ERROR, f"Unable to parse Dashboard Panel (data={panel}), ignored. {e}")) | ||
|
||
validation.append((Level.DEBUG, f'Defined {len(panelList)} dashboard panels')) | ||
return panelList, validation | ||
|
||
@classmethod | ||
def handleExperimentType(cls, defs: TestCaseData) -> [(Level, str)]: | ||
validation = [] | ||
if defs.Standard is None: | ||
defs.Standard = (defs.Custom is None) | ||
validation.append((Level.WARNING, f'Standard not defined, assuming {defs.Standard}. Keys: {defs.AllKeys}')) | ||
return validation | ||
|
||
@classmethod | ||
def createDashboard(cls, key: str, defs: TestCaseData) -> [(Level, str)]: | ||
validation = [] | ||
if defs.Dashboard is not None: | ||
cls.dashboards[key], validation = cls.getPanelList(defs.Dashboard) | ||
return validation | ||
|
||
@classmethod | ||
def createExtra(cls, key: str, defs: TestCaseData): | ||
cls.extra[key] = { | ||
'Standard': defs.Standard, | ||
'PublicCustom': (defs.Custom is not None and len(defs.Custom) == 0), | ||
'PrivateCustom': defs.Custom if defs.Custom is not None else [], | ||
'Parameters': defs.Parameters, | ||
'Distributed': defs.Distributed | ||
} | ||
|
||
@classmethod | ||
def validateParameters(cls, defs: TestCaseData) -> [(Level, str)]: | ||
validation = [] | ||
for name, info in defs.Parameters.items(): | ||
type, desc = (info['Type'], info['Description']) | ||
if name not in cls.parameters.keys(): | ||
cls.parameters[name] = (type, desc) | ||
else: | ||
oldType, oldDesc = cls.parameters[name] | ||
if type != oldType or desc != oldDesc: | ||
validation.append( | ||
(Level.WARNING, f"Redefined parameter '{name}' with different settings: " | ||
f"'{oldType}' - '{type}'; '{oldDesc}' - '{desc}'. " | ||
f"Cannot guarantee consistency.")) | ||
return validation | ||
|
||
@classmethod | ||
def ProcessData(cls, data: Dict) -> [(Level, str)]: | ||
version = str(data.pop('Version', 1)) | ||
|
||
match version: | ||
case '1': return cls.processV1Data(data) | ||
case '2': return cls.processV2Data(data) | ||
case _: raise RuntimeError(f"Unknown testcase version '{version}'.") | ||
|
||
@classmethod | ||
def processV1Data(cls, data: Dict) -> [(Level, str)]: | ||
validation = [] | ||
defs = TestCaseData(data) | ||
|
||
if defs.Dashboard is None: | ||
validation.append((Level.WARNING, f'Dashboard not defined. Keys: {defs.AllKeys}')) | ||
|
||
validation.extend( | ||
cls.handleExperimentType(defs)) | ||
|
||
keys = list(data.keys()) | ||
|
||
if len(keys) > 1: | ||
validation.append((Level.ERROR, f'Multiple TestCases defined on a single file: {list(keys)}')) | ||
|
||
for key in keys: | ||
cls.testCases[key], v = cls.GetActionList(data[key]) | ||
validation.extend(v) | ||
|
||
cls.createExtra(key, defs) | ||
|
||
validation.extend( | ||
cls.createDashboard(key, defs)) | ||
|
||
validation.extend( | ||
cls.validateParameters(defs)) | ||
|
||
return validation | ||
|
||
@classmethod | ||
def processV2Data(cls, data: Dict) -> [(Level, str)]: | ||
validation = [] | ||
defs = TestCaseData(data) | ||
|
||
validation.extend( | ||
cls.handleExperimentType(defs)) | ||
|
||
cls.testCases[defs.Name], v = cls.GetActionList(defs.Sequence) | ||
validation.extend(v) | ||
|
||
cls.createExtra(defs.Name, defs) | ||
|
||
validation.extend( | ||
cls.createDashboard(defs.Name, defs)) | ||
|
||
validation.extend( | ||
cls.validateParameters(defs)) | ||
|
||
return validation | ||
|
||
@classmethod | ||
def Clear(cls): | ||
cls.testCases = {} | ||
cls.extra = {} | ||
cls.dashboards = {} | ||
cls.parameters = {} | ||
|
||
@classmethod | ||
def GetCurrentTestCases(cls): | ||
return cls.testCases | ||
|
||
@classmethod | ||
def GetCurrentTestCaseExtras(cls): | ||
return cls.extra | ||
|
||
@classmethod | ||
def GetCurrentDashboards(cls): | ||
return cls.dashboards |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,33 @@ | ||
from Helper import Level | ||
from .loader_base import Loader | ||
from ..action_information import ActionInformation | ||
from typing import Dict, List | ||
|
||
|
||
class UeLoader(Loader): | ||
ues: Dict[str, List[ActionInformation]] = {} | ||
|
||
@classmethod | ||
def ProcessData(cls, data: Dict) -> [(Level, str)]: | ||
validation = [] | ||
keys = list(data.keys()) | ||
|
||
if len(keys) > 1: | ||
validation.append((Level.WARNING, f'Multiple UEs defined on a single file: {keys}')) | ||
|
||
for key in keys: | ||
if key in cls.ues.keys(): | ||
validation.append((Level.WARNING, f'Redefining UE {key}')) | ||
actions, v = cls.GetActionList(data[key]) | ||
validation.extend(v) | ||
cls.ues[key] = actions | ||
|
||
return validation | ||
|
||
@classmethod | ||
def Clear(cls): | ||
cls.ues = {} | ||
|
||
@classmethod | ||
def GetCurrentUEs(cls): | ||
return cls.ues |
Oops, something went wrong.