Skip to content
Projects
Groups
Snippets
Help
Loading...
Help
Support
Keyboard shortcuts
?
Submit feedback
Contribute to GitLab
Sign in / Register
Toggle navigation
W
wendelin-telecom
Project overview
Project overview
Details
Activity
Releases
Repository
Repository
Files
Commits
Branches
Tags
Contributors
Graph
Compare
Issues
0
Issues
0
List
Boards
Labels
Milestones
Merge Requests
0
Merge Requests
0
CI / CD
CI / CD
Pipelines
Jobs
Schedules
Analytics
Analytics
CI / CD
Repository
Value Stream
Wiki
Wiki
Snippets
Snippets
Members
Members
Collapse sidebar
Close sidebar
Activity
Graph
Create a new issue
Jobs
Commits
Issue Boards
Open sidebar
Paul Graydon
wendelin-telecom
Commits
84a046ea
Commit
84a046ea
authored
Aug 21, 2024
by
Paul Graydon
Browse files
Options
Browse Files
Download
Email Patches
Plain Diff
ors_wendelin_test: Clean up
parent
1ad709c4
Changes
1
Show whitespace changes
Inline
Side-by-side
Showing
1 changed file
with
10 additions
and
10 deletions
+10
-10
bt5/ors_wendelin_test/TestTemplateItem/portal_components/test.erp5.testORSWendelin.py
...mplateItem/portal_components/test.erp5.testORSWendelin.py
+10
-10
No files found.
bt5/ors_wendelin_test/TestTemplateItem/portal_components/test.erp5.testORSWendelin.py
View file @
84a046ea
...
...
@@ -43,7 +43,7 @@ class ORSWendelinTest(SecurityTestCase):
def
afterSetUp
(
self
):
# Set up variables for ORS ingestion testing
self
.
ors_
ingestion
=
self
.
portal
.
portal_ingestion_policies
.
ors
_ingestion
self
.
ors_
enb_log_ingestion
=
self
.
portal
.
portal_ingestion_policies
.
ors_enb_log
_ingestion
self
.
test_ors_example_log_valid
=
{
'log'
:
self
.
portal
.
web_page_module
.
test_ors_example_log_valid
.
getTextContent
()}
self
.
test_ors_example_log_invalid
=
{
'log'
:
self
.
portal
.
web_page_module
.
test_ors_example_log_invalid
.
getTextContent
()}
...
...
@@ -119,7 +119,7 @@ class ORSWendelinTest(SecurityTestCase):
data_supply
=
None
if
data_acquisition_unit
:
data_supply
=
data_acquisition_unit
.
DataAcquisitionUnit_createO
RS
DataSupply
(
batch
=
1
)
data_supply
=
data_acquisition_unit
.
DataAcquisitionUnit_createO
rs
DataSupply
(
batch
=
1
)
# Return all associated objects
return
{
...
...
@@ -134,7 +134,7 @@ class ORSWendelinTest(SecurityTestCase):
# Simulate a fluentd instance sending the given log data to Wendelin for ingestion
body
=
msgpack
.
packb
([
0
,
log_data
],
use_bin_type
=
True
)
env
=
{
'CONTENT_TYPE'
:
'application/octet-stream'
}
path
=
self
.
ors_ingestion
.
getPath
()
+
'/ingest?reference='
+
reference
path
=
self
.
ors_
enb_log_
ingestion
.
getPath
()
+
'/ingest?reference='
+
reference
publish_kw
=
dict
(
user
=
'ERP5TypeTestCase'
,
env
=
env
,
request_method
=
'POST'
,
stdin
=
StringIO
(
body
))
return
self
.
publish
(
path
,
**
publish_kw
)
...
...
@@ -168,7 +168,7 @@ class ORSWendelinTest(SecurityTestCase):
# Retrieve all items linked to the ingestion
data_acquisition_unit
=
self
.
portal
.
portal_catalog
.
getResultValue
(
portal_type
=
'Data Acquisition Unit'
,
reference
=
reference
)
data_supply
=
data_acquisition_unit
.
DataAcquisitionUnit_createO
RS
DataSupply
(
batch
=
1
)
data_supply
=
data_acquisition_unit
.
DataAcquisitionUnit_createO
rs
DataSupply
(
batch
=
1
)
self
.
tic
()
data_stream
=
self
.
_getDataStream
(
data_acquisition_unit
)
...
...
@@ -274,7 +274,7 @@ class ORSWendelinTest(SecurityTestCase):
self
.
tic
()
# Call the tested script which creates a related Data Supply
data_supply
=
data_acquisition_unit
.
DataAcquisitionUnit_createO
RS
DataSupply
(
batch
=
1
)
data_supply
=
data_acquisition_unit
.
DataAcquisitionUnit_createO
rs
DataSupply
(
batch
=
1
)
self
.
tic
()
# Check that the Data Supply exists and is validated
...
...
@@ -394,7 +394,7 @@ class ORSWendelinTest(SecurityTestCase):
self
.
assertTrue
(
data_acquisition_unit
is
not
None
)
# Check that the Data Supply exists and is validated
data_supply
=
data_acquisition_unit
.
DataAcquisitionUnit_createO
RS
DataSupply
(
batch
=
1
)
data_supply
=
data_acquisition_unit
.
DataAcquisitionUnit_createO
rs
DataSupply
(
batch
=
1
)
self
.
assertTrue
(
data_supply
is
not
None
)
# Generate new references
...
...
@@ -415,7 +415,7 @@ class ORSWendelinTest(SecurityTestCase):
self
.
assertTrue
(
new_data_acquisition_unit
is
not
None
)
# Check that the Data Supply exists and is validated
new_data_supply
=
new_data_acquisition_unit
.
DataAcquisitionUnit_createO
RS
DataSupply
(
batch
=
1
)
new_data_supply
=
new_data_acquisition_unit
.
DataAcquisitionUnit_createO
rs
DataSupply
(
batch
=
1
)
self
.
assertTrue
(
new_data_supply
is
not
None
)
# Add another ORS to the project, with the same reference and a new title
...
...
@@ -453,8 +453,8 @@ class ORSWendelinTest(SecurityTestCase):
self
.
assertTrue
(
all
(
data_array
is
not
None
for
data_array
in
ingestion_objects
[
'data_arrays'
]))
# Check that the data arrays containing the KPI data have correctly been initialized
self
.
assertTrue
(
ingestion_objects
[
'data_acquisition_unit'
].
Base_getER
AB
DataArrayKey
()
is
not
None
)
self
.
assertTrue
(
ingestion_objects
[
'data_acquisition_unit'
].
Base_getEU
TRAN
DataArrayKey
()
is
not
None
)
self
.
assertTrue
(
ingestion_objects
[
'data_acquisition_unit'
].
Base_getER
ab
DataArrayKey
()
is
not
None
)
self
.
assertTrue
(
ingestion_objects
[
'data_acquisition_unit'
].
Base_getEU
tran
DataArrayKey
()
is
not
None
)
if
valid_data
:
# If all data is valid: check that the arrays contain some data
...
...
@@ -506,7 +506,7 @@ class ORSWendelinTest(SecurityTestCase):
)
data_acquisition_unit
.
validate
()
self
.
tic
()
data_supply
=
data_acquisition_unit
.
DataAcquisitionUnit_createO
RS
DataSupply
(
batch
=
1
)
data_supply
=
data_acquisition_unit
.
DataAcquisitionUnit_createO
rs
DataSupply
(
batch
=
1
)
self
.
tic
()
# Force update on local roles to ensure proper permissions
...
...
Write
Preview
Markdown
is supported
0%
Try again
or
attach a new file
Attach a file
Cancel
You are about to add
0
people
to the discussion. Proceed with caution.
Finish editing this message first!
Cancel
Please
register
or
sign in
to comment