Skip to content

Commit

Permalink
Refactor test-infrastructure
Browse files Browse the repository at this point in the history
  • Loading branch information
christoph2 committed Oct 16, 2020
1 parent 223bbb1 commit d1e05d8
Show file tree
Hide file tree
Showing 5 changed files with 133 additions and 193 deletions.
Empty file removed tests/__init__.py
Empty file.
5 changes: 1 addition & 4 deletions tests/loader.py
Original file line number Diff line number Diff line change
Expand Up @@ -79,10 +79,7 @@ def loadFunctions(self):
setattr(self, fun.alias, inst)


class XCP(API):
FUNCTIONS = (
Function("Xcp_Init"),
)




Expand Down
1 change: 1 addition & 0 deletions tests/requirements.txt
Original file line number Diff line number Diff line change
@@ -1,3 +1,4 @@
pytest
hypothesis
mako
cpp-coveralls
309 changes: 123 additions & 186 deletions tests/test_xcpdaq.py
Original file line number Diff line number Diff line change
Expand Up @@ -7,94 +7,35 @@
from pprint import pprint
import pytest

def libname(name):
return os.path.abspath(os.path.join(os.path.dirname(__file__), name))
from xcp_if import XCP

from xcp_types import (
Xcp_ReturnType, XcpDaq_ListIntegerType, XcpDaq_ODTIntegerType, XcpDaq_ODTEntryIntegerType,
XcpDaq_ListConfigurationType, XcpDaq_ListStateType, XcpDaq_ODTEntryType, XcpDaq_EventType,
XcpDaq_ProcessorStateType, XcpDaq_ProcessorType, XcpDaq_MessageType, XcpDaq_EntityType,
XcpDaq_EntityKindType
)

FUNCS = """XcpDaq_Alloc
XcpDaq_AllocOdt
XcpDaq_AllocOdtEntry
XcpDaq_CopyMemory
XcpDaq_DequeueMessage
XcpDaq_EnqueueMessage
XcpDaq_Free
XcpDaq_GetEventConfiguration
XcpDaq_GetFirstPid
XcpDaq_GetListConfiguration
XcpDaq_GetListState
XcpDaq_GetOdtEntry
XcpDaq_GetProperties
XcpDaq_Init
XcpDaq_MainFunction
XcpDaq_SetProcessorState
XcpDaq_TriggerEvent
XcpDaq_ValidateList
XcpDaq_ValidateOdtEntry
XcpDaq_GetCounts
XcpDaq_GetDynamicEntities
XcpDaq_TotalDynamicEntityCount
Xcp_Init
XcpDaq_GetDynamicEntity
XcpDaq_GetDynamicEntities
XcpDaq_GetDtoBuffer
"""
def libname(name):
return os.path.abspath(os.path.join(os.path.dirname(__file__), name))


DLL_NAME = "./test_daq.so"

dll = ctypes.CDLL(DLL_NAME)

FUNCTIONS = {}
for func in FUNCS.splitlines():
func = func.strip()
FUNCTIONS[func] = getattr(dll, func)

@pytest.fixture
def daq():
xcp_init()
xcpdaq_init()

def xcp_init():
FUNCTIONS["Xcp_Init"]()

def xcpdaq_init():
FUNCTIONS["XcpDaq_Init"]()

def xcpdaq_get_counts():
entityCount = ctypes.c_uint16()
listCount = ctypes.c_uint16()
odtCount = ctypes.c_uint16()
FUNCTIONS["XcpDaq_GetCounts"](ctypes.byref(entityCount), ctypes.byref(listCount), ctypes.byref(odtCount))
return (entityCount.value, listCount.value, odtCount.value, )

def xcpdaq_free():
return FUNCTIONS["XcpDaq_Free"]()
def xcp():
xcp =XCP(dll)
xcp.Xcp_Init()
xcp.XcpDaq_Init()
yield xcp
del xcp

def xcpdaq_alloc(daq_count: int) -> int:
return FUNCTIONS["XcpDaq_Alloc"](daq_count)

def xcpdaq_allocodt(daq_list_num: int, odt_count: int) -> int:
return FUNCTIONS["XcpDaq_AllocOdt"](daq_list_num, odt_count)

def xcpdaq_allocodt_entry(daq_list_num: int, odt_num: int, entry_count: int) -> int:
return FUNCTIONS["XcpDaq_AllocOdtEntry"](daq_list_num, odt_num, entry_count)

def xcpdaq_total_dynamic_entity_count():
return FUNCTIONS["XcpDaq_TotalDynamicEntityCount"]()

def xcpdaq_get_dynamic_entities():
vp = FUNCTIONS["XcpDaq_GetDynamicEntities"]()
return vp

def xcpdaq_get_dto_buffer():
return FUNCTIONS["XcpDaq_GetDtoBuffer"]()

def xcpdaq_get_dynamic_entity(num):
vp = FUNCTIONS["XcpDaq_GetDynamicEntity"](num)
return vp

def xcpdaq_get_dynamic_entity_content(num):
addr = xcpdaq_get_dynamic_entity(num)
def xcpdaq_get_dynamic_entity_content(xcp, num):
addr = xcp.XcpDaq_GetDynamicEntity(num)
entry = ctypes.cast(addr, ctypes.POINTER(XcpDaq_EntityType))
kind = entry.contents.kind
entity = entry.contents.entity
Expand All @@ -110,129 +51,125 @@ def xcpdaq_get_dynamic_entity_content(num):
result = entity.odtEntry
return XcpDaq_EntityKindType(kind), result

def xcpdaq_check_entries_are_unused(start_idx: int = 0):
for idx in range(start_idx, xcpdaq_total_dynamic_entity_count()):
ent = xcpdaq_get_dynamic_entity_content(idx)
def xcpdaq_check_entries_are_unused(xcp, start_idx: int = 0):
for idx in range(start_idx, xcp.XcpDaq_TotalDynamicEntityCount()):
ent = xcpdaq_get_dynamic_entity_content(xcp, idx)
assert ent == (XcpDaq_EntityKindType.XCP_ENTITY_UNUSED, None)

##
## Test XcpDaq_AllocTransitionTable
##
def test_alloc_seq_err():
assert xcpdaq_alloc(1) == Xcp_ReturnType.ERR_SEQUENCE

def test_allocodt_seq_err(daq):
assert xcpdaq_allocodt(0, 2) == Xcp_ReturnType.ERR_SEQUENCE

def test_allocodt_entry_seq_err(daq):
assert xcpdaq_allocodt_entry(0, 1, 2) == Xcp_ReturnType.ERR_SEQUENCE

def test_alloc_ok(daq):
xcpdaq_check_entries_are_unused()
assert xcpdaq_alloc(5) == Xcp_ReturnType.ERR_SUCCESS
ec, _, _ = xcpdaq_get_counts()
for idx in range(0, ec):
state, ent = xcpdaq_get_dynamic_entity_content(idx)

print("ENT", state, ent.numOdts, ent.firstOdt, ent.mode)

xcpdaq_check_entries_are_unused(5)

def test_allocodt_ok(daq):
assert xcpdaq_alloc(1) == Xcp_ReturnType.ERR_SUCCESS
assert xcpdaq_allocodt(0, 2) == Xcp_ReturnType.ERR_SUCCESS

def test_allocodt_entry_ok(daq):
assert xcpdaq_alloc(1) == Xcp_ReturnType.ERR_SUCCESS
assert xcpdaq_allocodt(0, 2) == Xcp_ReturnType.ERR_SUCCESS
assert xcpdaq_allocodt_entry(0, 1, 1) == Xcp_ReturnType.ERR_SUCCESS
xcpdaq_get_counts()

def test_free_ok1(daq):
assert xcpdaq_alloc(1) == Xcp_ReturnType.ERR_SUCCESS
assert xcpdaq_free() == Xcp_ReturnType.ERR_SUCCESS

def test_free_ok2(daq):
assert xcpdaq_alloc(1) == Xcp_ReturnType.ERR_SUCCESS
assert xcpdaq_free() == Xcp_ReturnType.ERR_SUCCESS

def test_free_ok3(daq):
assert xcpdaq_alloc(1) == Xcp_ReturnType.ERR_SUCCESS
assert xcpdaq_allocodt(0, 1) == Xcp_ReturnType.ERR_SUCCESS
assert xcpdaq_free() == Xcp_ReturnType.ERR_SUCCESS

def test_free_ok4(daq):
assert xcpdaq_alloc(1) == Xcp_ReturnType.ERR_SUCCESS
assert xcpdaq_allocodt(0, 1) == Xcp_ReturnType.ERR_SUCCESS
assert xcpdaq_allocodt_entry(0, 1, 1) == Xcp_ReturnType.ERR_SUCCESS
assert xcpdaq_free() == Xcp_ReturnType.ERR_SUCCESS

def test_alloc_aftera_alloc(daq):
assert xcpdaq_alloc(1) == Xcp_ReturnType.ERR_SUCCESS
assert xcpdaq_alloc(2) == Xcp_ReturnType.ERR_SUCCESS

def test_allocodt_after_allocodt(daq):
assert xcpdaq_alloc(2) == Xcp_ReturnType.ERR_SUCCESS
assert xcpdaq_allocodt(0, 2) == Xcp_ReturnType.ERR_SUCCESS
assert xcpdaq_allocodt(1, 3) == Xcp_ReturnType.ERR_SUCCESS

def test_allocodt_entry_after_allocodt_entry(daq):
assert xcpdaq_alloc(1) == Xcp_ReturnType.ERR_SUCCESS
assert xcpdaq_allocodt(0, 2) == Xcp_ReturnType.ERR_SUCCESS
assert xcpdaq_allocodt_entry(0, 0, 1) == Xcp_ReturnType.ERR_SUCCESS
assert xcpdaq_allocodt_entry(0, 1, 2) == Xcp_ReturnType.ERR_SUCCESS

def test_alloc_after_allocodt(daq):
assert xcpdaq_alloc(2) == Xcp_ReturnType.ERR_SUCCESS
assert xcpdaq_allocodt(0, 2) == Xcp_ReturnType.ERR_SUCCESS
assert xcpdaq_alloc(3) == Xcp_ReturnType.ERR_SEQUENCE

def test_alloc_after_allocodt_entry(daq):
assert xcpdaq_alloc(1) == Xcp_ReturnType.ERR_SUCCESS
assert xcpdaq_allocodt(0, 2) == Xcp_ReturnType.ERR_SUCCESS
assert xcpdaq_allocodt_entry(0, 0, 1) == Xcp_ReturnType.ERR_SUCCESS
assert xcpdaq_alloc(3) == Xcp_ReturnType.ERR_SEQUENCE

def test_allocodt_after_allocodt_entry(daq):
assert xcpdaq_alloc(2) == Xcp_ReturnType.ERR_SUCCESS
assert xcpdaq_allocodt(0, 2) == Xcp_ReturnType.ERR_SUCCESS
assert xcpdaq_allocodt_entry(0, 0, 1) == Xcp_ReturnType.ERR_SUCCESS
assert xcpdaq_allocodt(1, 3) == Xcp_ReturnType.ERR_SEQUENCE

#for idx in range(8):
# ent = xcpdaq_get_dynamic_entity_content(idx)
# print(ent)
xcp =XCP(dll)
assert xcp.XcpDaq_Alloc(1) == Xcp_ReturnType.ERR_SEQUENCE

def test_allocodt_seq_err(xcp):
assert xcp.XcpDaq_AllocOdt(0, 2) == Xcp_ReturnType.ERR_SEQUENCE

def test_allocodt_entry_seq_err(xcp):
assert xcp.XcpDaq_AllocOdtEntry(0, 1, 2) == Xcp_ReturnType.ERR_SEQUENCE

def test_alloc_ok(xcp):
xcpdaq_check_entries_are_unused(xcp)
assert xcp.XcpDaq_Alloc(5) == Xcp_ReturnType.ERR_SUCCESS
#ec, _, _ = xcp.get_daq_counts()
#for idx in range(0, ec):
# state, ent = xcp.XcpDaq_TotalDynamicEntityCount()
# print("ENT", state, ent.numOdts, ent.firstOdt, ent.mode)

xcpdaq_check_entries_are_unused(xcp, 5)

def test_allocodt_ok(xcp):
assert xcp.XcpDaq_Alloc(1) == Xcp_ReturnType.ERR_SUCCESS
assert xcp.XcpDaq_AllocOdt(0, 2) == Xcp_ReturnType.ERR_SUCCESS

def test_allocodt_entry_ok(xcp):
assert xcp.XcpDaq_Alloc(1) == Xcp_ReturnType.ERR_SUCCESS
assert xcp.XcpDaq_AllocOdt(0, 2) == Xcp_ReturnType.ERR_SUCCESS
assert xcp.XcpDaq_AllocOdtEntry(0, 1, 1) == Xcp_ReturnType.ERR_SUCCESS

def test_free_ok1(xcp):
assert xcp.XcpDaq_Alloc(1) == Xcp_ReturnType.ERR_SUCCESS
assert xcp.XcpDaq_Free() == Xcp_ReturnType.ERR_SUCCESS

def test_free_ok2(xcp):
assert xcp.XcpDaq_Alloc(1) == Xcp_ReturnType.ERR_SUCCESS
assert xcp.XcpDaq_Free() == Xcp_ReturnType.ERR_SUCCESS

def test_free_ok3(xcp):
assert xcp.XcpDaq_Alloc(1) == Xcp_ReturnType.ERR_SUCCESS
assert xcp.XcpDaq_AllocOdt(0, 1) == Xcp_ReturnType.ERR_SUCCESS
assert xcp.XcpDaq_Free() == Xcp_ReturnType.ERR_SUCCESS

def test_free_ok4(xcp):
assert xcp.XcpDaq_Alloc(1) == Xcp_ReturnType.ERR_SUCCESS
assert xcp.XcpDaq_AllocOdt(0, 1) == Xcp_ReturnType.ERR_SUCCESS
assert xcp.XcpDaq_AllocOdtEntry(0, 1, 1) == Xcp_ReturnType.ERR_SUCCESS
assert xcp.XcpDaq_Free() == Xcp_ReturnType.ERR_SUCCESS

def test_alloc_aftera_alloc(xcp):
assert xcp.XcpDaq_Alloc(1) == Xcp_ReturnType.ERR_SUCCESS
assert xcp.XcpDaq_Alloc(2) == Xcp_ReturnType.ERR_SUCCESS

def test_allocodt_after_allocodt(xcp):
assert xcp.XcpDaq_Alloc(2) == Xcp_ReturnType.ERR_SUCCESS
assert xcp.XcpDaq_AllocOdt(0, 2) == Xcp_ReturnType.ERR_SUCCESS
assert xcp.XcpDaq_AllocOdt(1, 3) == Xcp_ReturnType.ERR_SUCCESS

def test_allocodt_entry_after_allocodt_entry(xcp):
assert xcp.XcpDaq_Alloc(1) == Xcp_ReturnType.ERR_SUCCESS
assert xcp.XcpDaq_AllocOdt(0, 2) == Xcp_ReturnType.ERR_SUCCESS
assert xcp.XcpDaq_AllocOdtEntry(0, 0, 1) == Xcp_ReturnType.ERR_SUCCESS
assert xcp.XcpDaq_AllocOdtEntry(0, 1, 2) == Xcp_ReturnType.ERR_SUCCESS

def test_alloc_after_allocodt(xcp):
assert xcp.XcpDaq_Alloc(2) == Xcp_ReturnType.ERR_SUCCESS
assert xcp.XcpDaq_AllocOdt(0, 2) == Xcp_ReturnType.ERR_SUCCESS
assert xcp.XcpDaq_Alloc(3) == Xcp_ReturnType.ERR_SEQUENCE

def test_alloc_after_allocodt_entry(xcp):
assert xcp.XcpDaq_Alloc(1) == Xcp_ReturnType.ERR_SUCCESS
assert xcp.XcpDaq_AllocOdt(0, 2) == Xcp_ReturnType.ERR_SUCCESS
assert xcp.XcpDaq_AllocOdtEntry(0, 0, 1) == Xcp_ReturnType.ERR_SUCCESS
assert xcp.XcpDaq_Alloc(3) == Xcp_ReturnType.ERR_SEQUENCE

def test_allocodt_after_allocodt_entry(xcp):
assert xcp.XcpDaq_Alloc(2) == Xcp_ReturnType.ERR_SUCCESS
assert xcp.XcpDaq_AllocOdt(0, 2) == Xcp_ReturnType.ERR_SUCCESS
assert xcp.XcpDaq_AllocOdtEntry(0, 0, 1) == Xcp_ReturnType.ERR_SUCCESS
assert xcp.XcpDaq_AllocOdt(1, 3) == Xcp_ReturnType.ERR_SEQUENCE


##
## Memory Overflows.
##
def test_alloc_out_of_mem(daq):
assert xcpdaq_alloc(101) == Xcp_ReturnType.ERR_MEMORY_OVERFLOW
def test_alloc_out_of_mem(xcp):
assert xcp.XcpDaq_Alloc(101) == Xcp_ReturnType.ERR_MEMORY_OVERFLOW

def test_allocodt_out_of_mem(daq):
assert xcpdaq_alloc(1) == Xcp_ReturnType.ERR_SUCCESS
assert xcpdaq_allocodt(0, 100) == Xcp_ReturnType.ERR_MEMORY_OVERFLOW
def test_allocodt_out_of_mem(xcp):
assert xcp.XcpDaq_Alloc(1) == Xcp_ReturnType.ERR_SUCCESS
assert xcp.XcpDaq_AllocOdt(0, 100) == Xcp_ReturnType.ERR_MEMORY_OVERFLOW

def test_allocodt_entry_out_of_mem(daq):
assert xcpdaq_alloc(1) == Xcp_ReturnType.ERR_SUCCESS
assert xcpdaq_allocodt(0, 2) == Xcp_ReturnType.ERR_SUCCESS
assert xcpdaq_allocodt_entry(0, 1, 100) == Xcp_ReturnType.ERR_MEMORY_OVERFLOW
def test_allocodt_entry_out_of_mem(xcp):
assert xcp.XcpDaq_Alloc(1) == Xcp_ReturnType.ERR_SUCCESS
assert xcp.XcpDaq_AllocOdt(0, 2) == Xcp_ReturnType.ERR_SUCCESS
assert xcp.XcpDaq_AllocOdtEntry(0, 1, 100) == Xcp_ReturnType.ERR_MEMORY_OVERFLOW

##
##
##
def test_basic_alloc1(daq):
assert xcpdaq_alloc(1) == Xcp_ReturnType.ERR_SUCCESS
assert xcpdaq_get_counts() == (1, 1, 0)

def test_basic_alloc2(daq):
assert xcpdaq_alloc(1) == Xcp_ReturnType.ERR_SUCCESS
assert xcpdaq_allocodt(0, 2) == Xcp_ReturnType.ERR_SUCCESS
assert xcpdaq_get_counts() == (3, 1, 2)

def test_basic_alloc3(daq):
assert xcpdaq_alloc(2) == Xcp_ReturnType.ERR_SUCCESS
assert xcpdaq_allocodt(0, 2) == Xcp_ReturnType.ERR_SUCCESS
assert xcpdaq_allocodt_entry(0, 0, 2) == Xcp_ReturnType.ERR_SUCCESS
assert xcpdaq_allocodt_entry(0, 1, 3) == Xcp_ReturnType.ERR_SUCCESS
assert xcpdaq_get_counts() == (9, 2, 2)
def test_basic_alloc1(xcp):
assert xcp.XcpDaq_Alloc(1) == Xcp_ReturnType.ERR_SUCCESS
assert xcp.get_daq_counts() == (1, 1, 0)

def test_basic_alloc2(xcp):
assert xcp.XcpDaq_Alloc(1) == Xcp_ReturnType.ERR_SUCCESS
assert xcp.XcpDaq_AllocOdt(0, 2) == Xcp_ReturnType.ERR_SUCCESS
assert xcp.get_daq_counts() == (3, 1, 2)

def test_basic_alloc3(xcp):
assert xcp.XcpDaq_Alloc(2) == Xcp_ReturnType.ERR_SUCCESS
assert xcp.XcpDaq_AllocOdt(0, 2) == Xcp_ReturnType.ERR_SUCCESS
assert xcp.XcpDaq_AllocOdtEntry(0, 0, 2) == Xcp_ReturnType.ERR_SUCCESS
assert xcp.XcpDaq_AllocOdtEntry(0, 1, 3) == Xcp_ReturnType.ERR_SUCCESS
assert xcp.get_daq_counts() == (9, 2, 2)
11 changes: 8 additions & 3 deletions tests/xcp.py → tests/xcp_if.py
Original file line number Diff line number Diff line change
Expand Up @@ -9,13 +9,12 @@
XcpDaq_ProcessorStateType, XcpDaq_ProcessorType, XcpDaq_MessageType, XcpDaq_EntityType
)


class XCP(API):
FUNCTIONS = (
Function("Xcp_Init"),
)

class DAQ(API):
FUNCTIONS = (

Function("XcpDaq_Init"),
Function("XcpDaq_Free", Xcp_ReturnType),
Function("XcpDaq_Alloc", Xcp_ReturnType, [XcpDaq_ListIntegerType]),
Expand Down Expand Up @@ -49,6 +48,12 @@ class DAQ(API):
#Function("", ),
)

def get_daq_counts(self):
entityCount = ctypes.c_uint16()
listCount = ctypes.c_uint16()
odtCount = ctypes.c_uint16()
self.XcpDaq_GetCounts(ctypes.byref(entityCount), ctypes.byref(listCount), ctypes.byref(odtCount))
return (entityCount.value, listCount.value, odtCount.value, )
"""
/*
** Predefined DAQ constants.
Expand Down

0 comments on commit d1e05d8

Please sign in to comment.