Skip to content
Projects
Groups
Snippets
Help
Loading...
Help
Support
Keyboard shortcuts
?
Submit feedback
Contribute to GitLab
Sign in / Register
Toggle navigation
S
slapos.core
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
Xiaowu Zhang
slapos.core
Commits
eb73d8d8
Commit
eb73d8d8
authored
Oct 03, 2012
by
Romain Courteaud
Browse files
Options
Browse Files
Download
Email Patches
Plain Diff
Dead code.
parent
dd275237
Changes
5
Hide whitespace changes
Inline
Side-by-side
Showing
5 changed files
with
39 additions
and
886 deletions
+39
-886
master/product/Vifib/tests/testVifibFiber.py
master/product/Vifib/tests/testVifibFiber.py
+0
-9
master/product/Vifib/tests/testVifibSlapBug.py
master/product/Vifib/tests/testVifibSlapBug.py
+0
-9
master/product/Vifib/tests/testVifibSlapComputerGetComputerPartitionList.py
...ib/tests/testVifibSlapComputerGetComputerPartitionList.py
+0
-13
master/product/Vifib/tests/testVifibSlapWebService.py
master/product/Vifib/tests/testVifibSlapWebService.py
+39
-831
master/product/Vifib/tests/testVifibSlapWebServiceSlaveInstance.py
...oduct/Vifib/tests/testVifibSlapWebServiceSlaveInstance.py
+0
-24
No files found.
master/product/Vifib/tests/testVifibFiber.py
View file @
eb73d8d8
...
...
@@ -110,15 +110,6 @@ class TestVifibFiberSubscription(testVifibSecurityMixin):
fiber_request
=
pending_request_list
[
0
]
sequence
.
edit
(
fiber_request_url
=
fiber_request
.
getRelativeUrl
())
def
stepStartFiberRequest
(
self
,
sequence
=
None
,
sequence_list
=
None
,
**
kw
):
"""Start the fiber request present in sequence"""
fiber_request_url
=
sequence
.
get
(
"fiber_request_url"
)
fiber_request
=
self
.
getPortal
().
restrictedTraverse
(
fiber_request_url
)
self
.
modifyFiberRequestState
(
"start_action"
,
sequence
,
fiber_request
)
self
.
assertEquals
(
fiber_request
.
getValidationState
(),
'started'
)
def
stepConfirmFiberRequest
(
self
,
sequence
=
None
,
sequence_list
=
None
,
**
kw
):
"""Confirm the fiber request present in sequence"""
fiber_request_url
=
sequence
.
get
(
"fiber_request_url"
)
...
...
master/product/Vifib/tests/testVifibSlapBug.py
View file @
eb73d8d8
...
...
@@ -135,9 +135,6 @@ class TestVifibSlapBug(TestVifibSlapWebServiceMixin):
def
stepRestoreSoftwareInstanceUidFromBufferA
(
self
,
sequence
,
**
kw
):
sequence
[
'software_instance_uid'
]
=
sequence
[
'buffer_a_software_instance_uid'
]
def
stepRestoreSoftwareInstanceUidFromBufferB
(
self
,
sequence
,
**
kw
):
sequence
[
'software_instance_uid'
]
=
sequence
[
'buffer_b_software_instance_uid'
]
def
stepRestoreComputerUidFromBufferA
(
self
,
sequence
,
**
kw
):
sequence
[
'computer_uid'
]
=
sequence
[
'buffer_a_computer_uid'
]
...
...
@@ -171,15 +168,9 @@ class TestVifibSlapBug(TestVifibSlapWebServiceMixin):
def
stepRestoreComputerPartitionUidFromBufferA
(
self
,
sequence
,
**
kw
):
sequence
[
'computer_partition_uid'
]
=
sequence
[
'buffer_a_computer_partition_uid'
]
def
stepRestoreComputerPartitionUidFromBufferB
(
self
,
sequence
,
**
kw
):
sequence
[
'computer_partition_uid'
]
=
sequence
[
'buffer_b_computer_partition_uid'
]
def
stepRestoreComputerPartitionReferenceFromBufferA
(
self
,
sequence
,
**
kw
):
sequence
[
'computer_partition_reference'
]
=
sequence
[
'buffer_a_computer_partition_reference'
]
def
stepRestoreComputerPartitionReferenceFromBufferB
(
self
,
sequence
,
**
kw
):
sequence
[
'computer_partition_reference'
]
=
sequence
[
'buffer_b_computer_partition_reference'
]
def
stepCheckHostingSubscriptionMultipleComputerAuditor
(
self
,
sequence
,
**
kw
):
hosting_subscription
=
self
.
portal
.
portal_catalog
.
getResultValue
(
uid
=
sequence
[
'hosting_subscription_uid'
])
...
...
master/product/Vifib/tests/testVifibSlapComputerGetComputerPartitionList.py
View file @
eb73d8d8
...
...
@@ -172,19 +172,6 @@ class TestVifibSlapComputerGetComputerPartitionList(TestVifibSlapWebServiceMixin
sequence_list
.
addSequenceString
(
sequence_string
)
sequence_list
.
play
(
self
)
def
stepDeliverInstanceSetupSalePackingList
(
self
,
sequence
,
**
kw
):
delivery
=
self
.
portal
.
portal_catalog
.
getResultValue
(
default_aggregate_uid
=
ComplexQuery
(
Query
(
default_aggregate_uid
=
sequence
[
'computer_partition_uid'
]),
Query
(
default_aggregate_uid
=
sequence
[
'software_instance_uid'
]),
operator
=
"AND"
),
portal_type
=
self
.
sale_packing_list_line_portal_type
,
simulation_state
=
'stopped'
,
resource_relative_url
=
self
.
portal
.
portal_preferences
\
.
getPreferredInstanceSetupResource
()
).
getParentValue
()
self
.
portal
.
portal_workflow
.
doActionFor
(
delivery
,
'deliver_action'
)
def
test_Computer_getComputerPartitionList_HostingResource_ConfirmedState
(
self
):
"""
Check that calling Computer.getComputerPartitionList works in
...
...
master/product/Vifib/tests/testVifibSlapWebService.py
View file @
eb73d8d8
...
...
@@ -251,9 +251,6 @@ class TestVifibSlapWebServiceMixin(testVifibMixin):
software_release_url
=
sequence
[
'software_release_uri'
],
free_for_request
=
1
)[
0
][
0
]
def
stepCheckSoftwareReleaseAvailableForRequest
(
self
,
sequence
,
**
kw
):
self
.
assertNotEqual
(
0
,
self
.
_getRequestBasedComputerPartitionCount
(
sequence
))
def
stepCheckSoftwareReleaseUnavailableForRequest
(
self
,
sequence
,
**
kw
):
self
.
assertEqual
(
0
,
self
.
_getRequestBasedComputerPartitionCount
(
sequence
))
...
...
@@ -267,9 +264,6 @@ class TestVifibSlapWebServiceMixin(testVifibMixin):
request
.
set
(
'portal_skin'
,
"View"
)
return
amount
def
stepCheckSoftwareReleaseInPublicTable
(
self
,
sequence
,
**
kw
):
self
.
assertNotEqual
(
0
,
self
.
_getSoftwareReleasePublicTableAmount
(
sequence
))
def
stepCheckSoftwareReleaseNotInPublicTable
(
self
,
sequence
,
**
kw
):
self
.
assertEqual
(
0
,
self
.
_getSoftwareReleasePublicTableAmount
(
sequence
))
...
...
@@ -344,23 +338,6 @@ class TestVifibSlapWebServiceMixin(testVifibMixin):
portal_type
=
'Sale Order Line'
).
getAggregateValue
(
portal_type
=
'Hosting Subscription'
).
getUid
())
def
stepSelectSoftwareInstanceFromCurrentSlaveInstance
(
self
,
sequence
):
slave_instance_reference
=
sequence
[
"software_instance_reference"
]
slave_instance
=
self
.
portal
.
portal_catalog
.
getResultValue
(
portal_type
=
self
.
slave_instance_portal_type
,
reference
=
slave_instance_reference
)
computer_partition
=
slave_instance
.
getAggregateRelatedValue
(
portal_type
=
"Sale Packing List Line"
).
getAggregateValue
(
portal_type
=
"Computer Partition"
)
software_instance
=
self
.
portal
.
portal_catalog
.
getResultValue
(
portal_type
=
"Sale Packing List Line"
,
aggregate_uid
=
computer_partition
.
getUid
(),
aggregatep_portal_type
=
self
.
software_instance_portal_type
,
).
getAggregateValue
(
portal_type
=
self
.
software_instance_portal_type
)
sequence
.
edit
(
software_instance_uid
=
software_instance
.
getUid
(),
software_instance_reference
=
software_instance
.
getReference
())
def
stepSetCurrentPersonSlapRequestedSoftwareInstance
(
self
,
sequence
,
**
kw
):
software_instance_list
=
[]
for
software_instance
in
self
.
portal
.
portal_catalog
(
...
...
@@ -464,12 +441,6 @@ class TestVifibSlapWebServiceMixin(testVifibMixin):
def
stepSetSoftwareTitleRandom
(
self
,
sequence
,
**
kw
):
sequence
[
'software_title'
]
=
str
(
random
())
def
stepCheckComputerPartitionInstanceCleanupSalePackingListDoesNotExists
(
self
,
sequence
,
**
kw
):
self
.
_checkComputerPartitionSalePackingListDoesNotExists
(
self
.
portal
.
portal_preferences
.
getPreferredInstanceCleanupResource
(),
sequence
)
def
stepCheckComputerPartitionInstanceCleanupSalePackingListConfirmed
(
self
,
sequence
,
**
kw
):
self
.
_checkComputerPartitionSalePackingListState
(
'confirmed'
,
...
...
@@ -518,13 +489,6 @@ class TestVifibSlapWebServiceMixin(testVifibMixin):
self
.
portal
.
portal_preferences
.
getPreferredInstanceHostingResource
(),
sequence
)
def
stepCheckComputerPartitionAccoutingResourceSalePackingListCancelled
(
self
,
sequence
,
**
kw
):
self
.
_checkComputerPartitionSalePackingListState
(
'cancelled'
,
self
.
portal
.
portal_catalog
.
getResultValue
(
uid
=
sequence
[
'service_uid'
]).
getRelativeUrl
(),
sequence
)
def
stepPersonRequestSlaveInstance
(
self
,
sequence
,
**
kw
):
kw
=
dict
(
instance_portal_type
=
self
.
slave_instance_portal_type
,
shared
=
True
,
...
...
@@ -641,23 +605,6 @@ class TestVifibSlapWebServiceMixin(testVifibMixin):
shared
=
shared
,
)
def
stepRequestSoftwareInstanceStartRaisesValueError
(
self
,
sequence
,
**
kw
):
instance
=
self
.
portal
.
portal_catalog
.
getResultValue
(
uid
=
sequence
[
'software_instance_uid'
])
if
instance
.
getPortalType
()
==
"Software Instance"
:
shared
=
False
elif
instance
.
getPortalType
()
==
"Slave Instance"
:
shared
=
True
else
:
raise
NotImplementedError
self
.
assertRaises
(
ValueError
,
instance
.
requestDestroy
,
software_release
=
instance
.
getRootSoftwareReleaseUrl
(),
instance_xml
=
instance
.
getTextContent
(),
software_type
=
instance
.
getSourceReference
(),
sla_xml
=
instance
.
getSlaXml
(),
shared
=
shared
,
)
def
stepRequestSoftwareInstanceStop
(
self
,
sequence
,
**
kw
):
instance
=
self
.
portal
.
portal_catalog
.
getResultValue
(
uid
=
sequence
[
'software_instance_uid'
])
...
...
@@ -1339,30 +1286,6 @@ class TestVifibSlapWebServiceMixin(testVifibMixin):
)
self
.
assertEquals
(
0
,
len
(
result_list
))
def
stepCheckDraftComputerState
(
self
,
sequence
,
**
kw
):
"""
Check that computer document is draft.
"""
computer_uid
=
sequence
[
"computer_uid"
]
computer
=
self
.
portal
.
portal_catalog
.
getResultValue
(
uid
=
computer_uid
)
self
.
assertEquals
(
"draft"
,
computer
.
getValidationState
())
def
stepValidateComputer
(
self
,
sequence
,
**
kw
):
"""
Validate the computer document.
"""
computer_uid
=
sequence
[
"computer_uid"
]
computer
=
self
.
portal
.
portal_catalog
.
getResultValue
(
uid
=
computer_uid
)
computer
.
validate
()
def
stepCheckValidatedComputerState
(
self
,
sequence
,
**
kw
):
"""
Check that computer document is validated.
"""
computer_uid
=
sequence
[
"computer_uid"
]
computer
=
self
.
portal
.
portal_catalog
.
getResultValue
(
uid
=
computer_uid
)
self
.
assertEquals
(
"validated"
,
computer
.
getValidationState
())
def
stepCheckUnauthorizedSlapGetComputerPartitionListCall
(
self
,
sequence
,
**
kw
):
computer_guid
=
sequence
.
get
(
"computer_reference"
,
str
(
random
()))
...
...
@@ -1393,30 +1316,6 @@ class TestVifibSlapWebServiceMixin(testVifibMixin):
self
.
assertRaises
(
slap
.
NotFoundError
,
self
.
slap
.
registerComputerPartition
,
computer_guid
,
partition_id
)
def
stepSelect0QuantityComputerPartition
(
self
,
sequence
,
**
kw
):
sequence
.
edit
(
computer_partition_quantity
=
0
)
def
stepSelectCurrentComputerPartitionAsSlaveOwner
(
self
,
sequence
,
**
kw
):
computer_partition_uid
=
sequence
[
"computer_partition_uid"
]
self
.
assertNotEqual
(
None
,
computer_partition_uid
)
sequence
.
edit
(
slave_owner_uid
=
computer_partition_uid
)
def
stepCreateComputerPartition
(
self
,
sequence
,
**
kw
):
"""
Create a computer partition document.
"""
computer_uid
=
sequence
[
"computer_uid"
]
partition_reference
=
sequence
[
"computer_partition_reference"
]
computer
=
self
.
portal
.
portal_catalog
.
getResultValue
(
uid
=
computer_uid
)
computer_partition
=
computer
.
newContent
(
portal_type
=
self
.
computer_partition_portal_type
,
reference
=
partition_reference
)
self
.
markManualCreation
(
computer_partition
)
# Mark newly created computer partition as free by default
computer_partition
.
markFree
()
sequence
.
edit
(
computer_partition_uid
=
computer_partition
.
getUid
())
def
stepCheckSuccessSlapRegisterComputerPartitionCall
(
self
,
sequence
,
**
kw
):
"""
Check that slap.registerComputerPartition is successfully called.
...
...
@@ -1527,17 +1426,6 @@ class TestVifibSlapWebServiceMixin(testVifibMixin):
self
.
markManualCreation
(
software_release
)
sequence
.
edit
(
software_release_uid
=
software_release
.
getUid
())
def
stepCheckUnexistingSoftwareRelease
(
self
,
sequence
,
**
kw
):
"""
Check that it is not present in the system.
"""
url
=
sequence
[
"software_release_uri"
]
result_list
=
self
.
portal
.
portal_catalog
(
portal_type
=
self
.
software_release_portal_type
,
url_string
=
url
,
)
self
.
assertEquals
(
0
,
len
(
result_list
))
def
stepSetSoftwareProductToSoftwareRelease
(
self
,
sequence
,
**
kw
):
"""
Associate software product to software release
...
...
@@ -1562,15 +1450,6 @@ class TestVifibSlapWebServiceMixin(testVifibMixin):
software_release
.
portal_workflow
.
doActionFor
(
software_release
,
'submit_action'
)
def
stepPublishSoftwareRelease
(
self
,
sequence
,
**
kw
):
"""
Validate the software release document.
"""
software_release_uid
=
sequence
[
"software_release_uid"
]
software_release
=
self
.
portal
.
portal_catalog
.
getResultValue
(
uid
=
software_release_uid
)
software_release
.
publish
()
def
stepPublishByActionSoftwareRelease
(
self
,
sequence
,
**
kw
):
"""
Validate the software release document.
...
...
@@ -1581,40 +1460,9 @@ class TestVifibSlapWebServiceMixin(testVifibMixin):
software_release
.
portal_workflow
.
doActionFor
(
software_release
,
'publish_action'
)
def
stepCheckDraftSoftwareReleaseState
(
self
,
sequence
,
**
kw
):
"""
Check that software release is draft.
"""
software_release_uid
=
sequence
[
"software_release_uid"
]
software_release
=
self
.
portal
.
portal_catalog
.
getResultValue
(
uid
=
software_release_uid
)
self
.
assertEquals
(
"draft"
,
software_release
.
getValidationState
())
def
stepCheckPublishedSoftwareReleaseState
(
self
,
sequence
,
**
kw
):
"""
Check that software release is validated.
"""
software_release_uid
=
sequence
[
"software_release_uid"
]
software_release
=
self
.
portal
.
portal_catalog
.
getResultValue
(
uid
=
software_release_uid
)
self
.
assertEquals
(
"published"
,
software_release
.
getValidationState
())
def
stepCheckNotFoundComputerGetComputerPartitionCall
(
self
,
sequence
,
**
kw
):
"""
Check that Computer.getComputerPartitionList is successfully called.
"""
computer_guid
=
sequence
[
"computer_reference"
]
self
.
slap
=
slap
.
slap
()
self
.
slap
.
initializeConnection
(
self
.
server_url
,
timeout
=
None
)
computer
=
self
.
slap
.
registerComputer
(
computer_guid
)
self
.
assertRaises
(
slap
.
NotFoundError
,
computer
.
getComputerPartitionList
)
def
stepSelectYetAnotherRequestedReference
(
self
,
sequence
,
**
kw
):
sequence
.
edit
(
requested_reference
=
'yet_another_requested_reference'
)
def
stepSelectYetAnotherRequestedSoftwareType
(
self
,
sequence
,
**
kw
):
sequence
.
edit
(
requested_software_type
=
'yet_another_requested_software_type'
)
def
stepSelectAnotherRequestedReference
(
self
,
sequence
,
**
kw
):
sequence
.
edit
(
requested_reference
=
'another_requested_reference'
)
...
...
@@ -1862,144 +1710,6 @@ class TestVifibSlapWebServiceMixin(testVifibMixin):
sequence
.
edit
(
software_instance_uid
=
root_software_instance_uid
,
computer_partition_reference
=
computer_partition_reference
)
def
stepRequestComputerPartitionDifferentReferenceSameTransaction
(
self
,
sequence
,
**
kw
):
software_release_uri
=
sequence
[
'software_release_uri'
]
computer_guid
=
sequence
[
"computer_reference"
]
requested_parameter_dict
=
sequence
[
'requested_parameter_dict'
]
software_instance_uid
=
sequence
[
'software_instance_uid'
]
software_instance
=
self
.
portal
.
portal_catalog
.
getResultValue
(
uid
=
software_instance_uid
)
# Slap Tool have to be used directly to simulate same transaction
raise
NotImplementedError
(
"Do not test slap tool but slap library"
)
children_software_instance
=
software_instance
\
.
SoftwareInstance_findAlreadyExisting
(
'children_a'
)
slap_tool_response
=
self
.
portal
.
portal_slap
.
requestComputerPartition
(
computer_guid
,
children_software_instance
\
.
SoftwareInstance_getComputerPartition
().
getReference
(),
software_release_uri
,
'children_a_child'
,
requested_parameter_dict
)
self
.
assertEqual
(
408
,
slap_tool_response
.
getStatus
())
children_software_instance
=
software_instance
\
.
SoftwareInstance_findAlreadyExisting
(
'children_b'
)
slap_tool_response
=
self
.
portal
.
portal_slap
.
requestComputerPartition
(
computer_guid
,
children_software_instance
\
.
SoftwareInstance_getComputerPartition
().
getReference
(),
software_release_uri
,
'children_b_child'
,
requested_parameter_dict
)
self
.
assertEqual
(
408
,
slap_tool_response
.
getStatus
())
def
stepRequestComputerPartitionSameReferenceSameTransaction
(
self
,
sequence
,
**
kw
):
raise
NotImplementedError
(
'Wrong logic'
)
software_release_uri
=
sequence
[
'software_release_uri'
]
requested_reference
=
sequence
[
'requested_reference'
]
computer_guid
=
sequence
[
"computer_reference"
]
requested_parameter_dict
=
sequence
[
'requested_parameter_dict'
]
software_instance_uid
=
sequence
[
'software_instance_uid'
]
software_instance
=
self
.
portal
.
portal_catalog
.
getResultValue
(
uid
=
software_instance_uid
)
self
.
slap
=
slap
.
slap
()
# Slap Tool have to be used directly to simulate same transaction
children_software_instance
=
\
software_instance
.
portal_catalog
.
getResultValue
(
portal_type
=
"Software Instance"
,
source_reference
=
'children_a'
,
root_uid
=
software_instance_uid
)
movement
=
children_software_instance
.
getAggregateRelatedValue
(
portal_type
=
"Sale Packing List Line"
)
children_partition
=
movement
.
getAggregateValue
(
portal_type
=
"Computer Partition"
)
slap_computer_partition
=
self
.
slap
.
registerComputerPartition
(
computer_guid
,
children_partition
.
getReference
())
slap_computer_partition
.
request
(
software_release
=
software_release_uri
,
software_type
=
requested_reference
,
partition_reference
=
requested_reference
,
partition_parameter_kw
=
requested_parameter_dict
)
children_software_instance
=
\
software_instance
.
portal_catalog
.
getResultValue
(
portal_type
=
"Software Instance"
,
source_reference
=
'children_b'
,
root_uid
=
software_instance_uid
)
movement
=
children_software_instance
.
getAggregateRelatedValue
(
portal_type
=
"Sale Packing List Line"
)
children_partition
=
movement
.
getAggregateValue
(
portal_type
=
"Computer Partition"
)
slap_computer_partition
=
self
.
slap
.
registerComputerPartition
(
computer_guid
,
children_partition
.
getReference
())
slap_computer_partition
.
request
(
software_release
=
software_release_uri
,
software_type
=
requested_reference
,
partition_reference
=
requested_reference
,
partition_parameter_kw
=
requested_parameter_dict
)
def
stepRequestSoftwareInstanceStartCheckSerializeIsCalled
(
self
,
sequence
):
# check that on being_requested serialise is being called
# code stolen from testERP5Security:test_MultiplePersonReferenceConcurrentTransaction
class
DummyTestException
(
Exception
):
pass
def
verify_serialize_call
(
self
):
# it is checking that anything below computer_module raises exception
# thanks to this this test do not have to be destructive
if
self
.
getPortalType
()
==
"Software Instance"
:
raise
DummyTestException
else
:
return
self
.
serialize_call
()
from
Products.ERP5Type.Base
import
Base
Base
.
serialize_call
=
Base
.
serialize
Base
.
serialize
=
verify_serialize_call
try
:
self
.
assertRaises
(
DummyTestException
,
self
.
portal
.
portal_catalog
.
getResultValue
(
uid
=
sequence
[
'software_instance_uid'
]).
requestStartComputerPartition
)
finally
:
Base
.
serialize
=
Base
.
serialize_call
def
stepRequestSoftwareInstanceDestroyCheckSerializeIsCalled
(
self
,
sequence
):
# check that on being_requested serialise is being called
# code stolen from testERP5Security:test_MultiplePersonReferenceConcurrentTransaction
class
DummyTestException
(
Exception
):
pass
def
verify_serialize_call
(
self
):
# it is checking that anything below computer_module raises exception
# thanks to this this test do not have to be destructive
if
self
.
getPortalType
()
==
"Software Instance"
:
raise
DummyTestException
else
:
return
self
.
serialize_call
()
from
Products.ERP5Type.Base
import
Base
Base
.
serialize_call
=
Base
.
serialize
Base
.
serialize
=
verify_serialize_call
instance
=
self
.
portal
.
portal_catalog
.
getResultValue
(
uid
=
sequence
[
'software_instance_uid'
])
if
instance
.
getPortalType
()
==
"Software Instance"
:
shared
=
False
elif
instance
.
getPortalType
()
==
"Slave Instance"
:
shared
=
True
else
:
raise
NotImplementedError
try
:
self
.
assertRaises
(
DummyTestException
,
instance
.
requestDestroy
,
software_release
=
instance
.
getRootSoftwareReleaseUrl
(),
instance_xml
=
instance
.
getTextContent
(),
software_type
=
instance
.
getSourceReference
(),
sla_xml
=
instance
.
getSlaXml
(),
shared
=
shared
,
)
finally
:
Base
.
serialize
=
Base
.
serialize_call
def
stepConfirmSaleOrderOrderedToCheckSerializeCalledOnSelected
(
self
,
sequence
,
**
kw
):
...
...
@@ -2222,17 +1932,6 @@ class TestVifibSlapWebServiceMixin(testVifibMixin):
computer
=
self
.
slap
.
registerComputer
(
computer_guid
)
self
.
assertRaises
(
slap
.
Unauthorized
,
computer
.
getSoftwareReleaseList
)
def
stepCheckNotFoundComputerGetSoftwareReleaseListCall
(
self
,
sequence
,
**
kw
):
"""
Check that Computer.getSoftwareReleaseList raise NotFoundError is the
computer doesn't exist.
"""
computer_guid
=
sequence
[
"computer_reference"
]
self
.
slap
=
slap
.
slap
()
self
.
slap
.
initializeConnection
(
self
.
server_url
,
timeout
=
None
)
computer
=
self
.
slap
.
registerComputer
(
computer_guid
)
self
.
assertRaises
(
slap
.
NotFoundError
,
computer
.
getSoftwareReleaseList
)
def
stepCheckEmptyComputerGetSoftwareReleaseListCall
(
self
,
sequence
,
**
kw
):
"""
Check that Computer.getSoftwareReleaseList returns an empty list.
...
...
@@ -2283,19 +1982,7 @@ class TestVifibSlapWebServiceMixin(testVifibMixin):
self
.
assertRaises
(
slap
.
NotFoundError
,
software_release
.
building
)
def
stepCheckSuccessSoftwareReleaseBuildingCall
(
self
,
sequence
,
**
kw
):
"""
Check that calling SoftwareRelease.building works.
"""
computer_guid
=
sequence
[
"computer_reference"
]
self
.
slap
=
slap
.
slap
()
self
.
slap
.
initializeConnection
(
self
.
server_url
,
timeout
=
None
)
computer
=
self
.
slap
.
registerComputer
(
computer_guid
)
software_release
=
computer
.
getSoftwareReleaseList
()[
0
]
software_release
.
building
()
def
stepCheckNotFoundSoftwareReleaseBuildingCall
(
self
,
sequence
,
**
kw
):
def
stepCheckNotFoundSoftwareReleaseDestroyedCall
(
self
,
sequence
,
**
kw
):
computer_guid
=
sequence
[
"computer_reference"
]
self
.
slap
=
slap
.
slap
()
self
.
slap
.
initializeConnection
(
self
.
server_url
,
timeout
=
None
)
...
...
@@ -2303,349 +1990,66 @@ class TestVifibSlapWebServiceMixin(testVifibMixin):
sequence
[
'software_release_uri'
])
# Note: Hackish
software_release
.
_computer_guid
=
computer_guid
self
.
assertRaises
(
slap
.
NotFoundError
,
software_release
.
building
)
self
.
assertRaises
(
slap
.
NotFoundError
,
software_release
.
destroyed
)
def
stepCheckTicAndNotFoundSoftwareReleaseBuildingCall
(
self
,
sequence
,
**
kw
):
"""
Check that calling SoftwareRelease.building raises NotFoundError
after using tic.
"""
computer_guid
=
sequence
[
"computer_reference"
]
def
stepCheckNotFoundSoftwareReleaseDestroyedAfterRegisterCall
(
self
,
sequence
,
**
kw
):
url
=
sequence
[
"software_release_uri"
]
self
.
slap
=
slap
.
slap
()
self
.
slap
.
initializeConnection
(
self
.
server_url
,
timeout
=
None
)
computer
=
self
.
slap
.
registerComputer
(
computer_guid
)
software_release
=
computer
.
getSoftwareReleaseList
()[
0
]
self
.
tic
()
transaction
.
commit
()
software_release
=
self
.
slap
.
registerSoftwareRelease
(
url
)
self
.
assertRaises
(
slap
.
NotFoundError
,
software_release
.
building
)
self
.
assertRaises
(
slap
.
NotFoundError
,
software_release
.
destroyed
)
def
stepCheckNotFoundSoftwareReleaseAvailableCall
(
self
,
sequence
,
**
kw
):
def
stepCheckNotFoundSoftwareReleaseAvailableAfterRegisterCall
(
self
,
sequence
,
**
kw
):
"""
Check that calling SoftwareRelease.available
raises NotFound
after using tic.
Check that calling SoftwareRelease.available
after just registration raises a
NotFoundError
"""
computer_guid
=
sequence
[
"computer_reference
"
]
url
=
sequence
[
"software_release_uri
"
]
self
.
slap
=
slap
.
slap
()
self
.
slap
.
initializeConnection
(
self
.
server_url
,
timeout
=
None
)
software_release
=
self
.
slap
.
registerSoftwareRelease
(
sequence
[
'software_release_uri'
])
# Note: Hackish
software_release
.
_computer_guid
=
computer_guid
software_release
=
self
.
slap
.
registerSoftwareRelease
(
url
)
self
.
assertRaises
(
slap
.
NotFoundError
,
software_release
.
available
)
def
stepCheckNotFoundSoftwareReleaseDestroyedCall
(
self
,
sequence
,
**
kw
):
computer_guid
=
sequence
[
"computer_reference"
]
def
stepCheckNotFoundSoftwareReleaseErrorAfterRegisterCall
(
self
,
sequence
,
**
kw
):
"""
Check that calling SoftwareRelease.error after just registration raises a
NotFoundError
"""
url
=
sequence
[
"software_release_uri"
]
self
.
slap
=
slap
.
slap
()
self
.
slap
.
initializeConnection
(
self
.
server_url
,
timeout
=
None
)
software_release
=
self
.
slap
.
registerSoftwareRelease
(
sequence
[
'software_release_uri'
])
# Note: Hackish
software_release
.
_computer_guid
=
computer_guid
self
.
assertRaises
(
slap
.
NotFoundError
,
software_release
.
destroyed
)
software_release
=
self
.
slap
.
registerSoftwareRelease
(
url
)
self
.
assertRaises
(
slap
.
NotFoundError
,
software_release
.
error
,
"ErrorLog"
)
def
stepCheckTicAndNotFoundSoftwareReleaseAvailableCall
(
self
,
sequence
,
**
kw
):
def
stepCheckNotFoundComputerPartitionBuildingAfterRegisterCall
(
self
,
sequence
,
**
kw
):
"""
Check that calling
SoftwareRelease.available raises NotFound
after using tic.
Check that calling
ComputerPartition.building after just registrationraises a
NotFoundError
"""
computer_guid
=
sequence
[
"computer_reference"
]
partition_id
=
sequence
[
"computer_partition_reference_list"
][
0
]
self
.
slap
=
slap
.
slap
()
self
.
slap
.
initializeConnection
(
self
.
server_url
,
timeout
=
None
)
computer
=
self
.
slap
.
registerComputer
(
computer_guid
)
software_release
=
computer
.
getSoftwareReleaseList
()[
0
]
self
.
tic
()
transaction
.
commit
()
computer_partition
=
self
.
slap
.
registerComputerPartition
(
computer_guid
,
partition_id
)
self
.
assertRaises
(
slap
.
NotFoundError
,
software_release
.
available
)
self
.
assertRaises
(
slap
.
NotFoundError
,
computer_partition
.
building
)
def
stepCheckSoftwareReleaseErrorCall
(
self
,
sequence
,
**
kw
):
def
stepCheckNotFoundComputerPartitionAvailableAfterRegisterCall
(
self
,
sequence
,
**
kw
):
"""
Check that calling ComputerPartition.available after just registration
raises a NotFoundError
"""
computer_guid
=
sequence
[
"computer_reference"
]
self
.
slap
=
slap
.
slap
()
self
.
slap
.
initializeConnection
(
self
.
server_url
,
timeout
=
None
)
software_release
=
self
.
slap
.
registerSoftwareRelease
(
sequence
[
'software_release_uri'
])
# Note: Hackish
software_release
.
_computer_guid
=
computer_guid
software_release
.
error
(
"ErrorLog"
)
def
stepCheckTicAndSoftwareReleaseErrorCall
(
self
,
sequence
,
**
kw
):
"""
Check that calling SoftwareRelease.error raises NotFound
after using tic.
"""
computer_guid
=
sequence
[
"computer_reference"
]
self
.
slap
=
slap
.
slap
()
self
.
slap
.
initializeConnection
(
self
.
server_url
,
timeout
=
None
)
computer
=
self
.
slap
.
registerComputer
(
computer_guid
)
software_release
=
computer
.
getSoftwareReleaseList
()[
0
]
self
.
tic
()
transaction
.
commit
()
software_release
.
error
(
"ErrorLog"
)
def
stepCheckTicAndNotFoundSoftwareReleaseErrorCall
(
self
,
sequence
,
**
kw
):
"""
Check that calling SoftwareRelease.error raises NotFound
after using tic.
"""
computer_guid
=
sequence
[
"computer_reference"
]
self
.
slap
=
slap
.
slap
()
self
.
slap
.
initializeConnection
(
self
.
server_url
,
timeout
=
None
)
computer
=
self
.
slap
.
registerComputer
(
computer_guid
)
software_release
=
computer
.
getSoftwareReleaseList
()[
0
]
self
.
tic
()
transaction
.
commit
()
self
.
assertRaises
(
slap
.
NotFoundError
,
software_release
.
error
,
"ErrorLog"
)
def
stepCheckNotFoundSoftwareReleaseDestroyedAfterRegisterCall
(
self
,
sequence
,
**
kw
):
url
=
sequence
[
"software_release_uri"
]
self
.
slap
=
slap
.
slap
()
self
.
slap
.
initializeConnection
(
self
.
server_url
,
timeout
=
None
)
software_release
=
self
.
slap
.
registerSoftwareRelease
(
url
)
self
.
assertRaises
(
slap
.
NotFoundError
,
software_release
.
destroyed
)
def
stepCheckNotFoundSoftwareReleaseAvailableAfterRegisterCall
(
self
,
sequence
,
**
kw
):
"""
Check that calling SoftwareRelease.available after just registration raises a
NotFoundError
"""
url
=
sequence
[
"software_release_uri"
]
self
.
slap
=
slap
.
slap
()
self
.
slap
.
initializeConnection
(
self
.
server_url
,
timeout
=
None
)
software_release
=
self
.
slap
.
registerSoftwareRelease
(
url
)
self
.
assertRaises
(
slap
.
NotFoundError
,
software_release
.
available
)
def
stepCheckSuccessSoftwareReleaseAvailableCall
(
self
,
sequence
,
**
kw
):
"""
Check that calling SoftwareRelease.available works.
"""
computer_guid
=
sequence
[
"computer_reference"
]
self
.
slap
=
slap
.
slap
()
self
.
slap
.
initializeConnection
(
self
.
server_url
,
timeout
=
None
)
computer
=
self
.
slap
.
registerComputer
(
computer_guid
)
software_release
=
computer
.
getSoftwareReleaseList
()[
0
]
software_release
.
available
()
def
stepCheckSuccessSoftwareReleaseDestroyedCall
(
self
,
sequence
,
**
kw
):
computer_guid
=
sequence
[
"computer_reference"
]
self
.
slap
=
slap
.
slap
()
self
.
slap
.
initializeConnection
(
self
.
server_url
,
timeout
=
None
)
computer
=
self
.
slap
.
registerComputer
(
computer_guid
)
software_release
=
computer
.
getSoftwareReleaseList
()[
0
]
software_release
.
destroyed
()
def
stepCheckUnauthorizedSoftwareReleaseAvailableCall
(
self
,
sequence
,
**
kw
):
"""
Check that calling SoftwareRelease.available is Unauthorized
"""
computer_guid
=
sequence
[
"computer_reference"
]
self
.
slap
=
slap
.
slap
()
self
.
slap
.
initializeConnection
(
self
.
server_url
,
timeout
=
None
)
computer
=
self
.
slap
.
registerComputer
(
computer_guid
)
software_release
=
computer
.
getSoftwareReleaseList
()[
0
]
self
.
assertRaises
(
slap
.
Unauthorized
,
software_release
.
available
)
def
stepCheckNotFoundSoftwareReleaseErrorAfterRegisterCall
(
self
,
sequence
,
**
kw
):
"""
Check that calling SoftwareRelease.error after just registration raises a
NotFoundError
"""
url
=
sequence
[
"software_release_uri"
]
self
.
slap
=
slap
.
slap
()
self
.
slap
.
initializeConnection
(
self
.
server_url
,
timeout
=
None
)
software_release
=
self
.
slap
.
registerSoftwareRelease
(
url
)
self
.
assertRaises
(
slap
.
NotFoundError
,
software_release
.
error
,
"ErrorLog"
)
def
stepCheckSuccessSoftwareReleaseErrorCall
(
self
,
sequence
,
**
kw
):
"""
Check that calling SoftwareRelease.error works.
"""
computer_guid
=
sequence
[
"computer_reference"
]
self
.
slap
=
slap
.
slap
()
self
.
slap
.
initializeConnection
(
self
.
server_url
,
timeout
=
None
)
computer
=
self
.
slap
.
registerComputer
(
computer_guid
)
software_release
=
computer
.
getSoftwareReleaseList
()[
0
]
software_release
.
error
(
"ErrorLog"
)
def
stepCheckUnauthorizedSoftwareReleaseBuildingCall
(
self
,
sequence
,
**
kw
):
"""
Check that calling SoftwareRelease.building raises an Unauthorized error
"""
computer_guid
=
sequence
[
"computer_reference"
]
self
.
slap
=
slap
.
slap
()
self
.
slap
.
initializeConnection
(
self
.
server_url
,
timeout
=
None
)
computer
=
self
.
slap
.
registerComputer
(
computer_guid
)
software_release
=
computer
.
getSoftwareReleaseList
()[
0
]
self
.
assertRaises
(
slap
.
Unauthorized
,
software_release
.
building
)
def
stepCheckTicAndSuccessSoftwareReleaseErrorCall
(
self
,
sequence
,
**
kw
):
"""
Check that calling SoftwareRelease.building is OK
after using tic.
"""
computer_guid
=
sequence
[
"computer_reference"
]
self
.
slap
=
slap
.
slap
()
self
.
slap
.
initializeConnection
(
self
.
server_url
,
timeout
=
None
)
computer
=
self
.
slap
.
registerComputer
(
computer_guid
)
software_release
=
computer
.
getSoftwareReleaseList
()[
0
]
self
.
tic
()
transaction
.
commit
()
software_release
.
error
(
"ErrorLog"
)
def
stepCheckSuccessComputerPartitionBuildingCall
(
self
,
sequence
,
**
kw
):
"""
Check that calling ComputerPartition.building works
"""
computer_guid
=
sequence
[
"computer_reference"
]
partition_id
=
sequence
[
"computer_partition_reference"
]
self
.
slap
=
slap
.
slap
()
self
.
slap
.
initializeConnection
(
self
.
server_url
,
timeout
=
None
)
computer_partition
=
self
.
slap
.
registerComputerPartition
(
computer_guid
,
partition_id
)
computer_partition
.
building
()
def
stepCheckNotFoundComputerPartitionBuildingCall
(
self
,
sequence
,
**
kw
):
"""
Check that calling ComputerPartition.building after just registration raises a
NotFound
"""
computer_guid
=
sequence
[
"computer_reference"
]
partition_id
=
sequence
[
"computer_partition_reference"
]
self
.
slap
=
slap
.
slap
()
self
.
slap
.
initializeConnection
(
self
.
server_url
,
timeout
=
None
)
computer_partition
=
self
.
slap
.
registerComputerPartition
(
computer_guid
,
partition_id
)
self
.
assertRaises
(
slap
.
NotFoundError
,
computer_partition
.
building
)
def
stepCheckUnauthorizedComputerPartitionBuildingCall
(
self
,
sequence
,
**
kw
):
"""
Check that calling ComputerPartition.building after just registration raises a
Unauthorized
"""
computer_guid
=
sequence
[
"computer_reference"
]
partition_id
=
sequence
[
"computer_partition_reference"
]
self
.
slap
=
slap
.
slap
()
self
.
slap
.
initializeConnection
(
self
.
server_url
,
timeout
=
None
)
computer_partition
=
self
.
slap
.
registerComputerPartition
(
computer_guid
,
partition_id
)
self
.
assertRaises
(
slap
.
Unauthorized
,
computer_partition
.
building
)
def
stepCheckNotFoundComputerPartitionBuildingAfterRegisterCall
(
self
,
sequence
,
**
kw
):
"""
Check that calling ComputerPartition.building after just registrationraises a
NotFoundError
"""
computer_guid
=
sequence
[
"computer_reference"
]
partition_id
=
sequence
[
"computer_partition_reference_list"
][
0
]
self
.
slap
=
slap
.
slap
()
self
.
slap
.
initializeConnection
(
self
.
server_url
,
timeout
=
None
)
computer_partition
=
self
.
slap
.
registerComputerPartition
(
computer_guid
,
partition_id
)
self
.
assertRaises
(
slap
.
NotFoundError
,
computer_partition
.
building
)
def
stepAssertConfirmSalePackingListRaisesUnsupportedWorkflowMethod
(
self
,
sequence
,
**
kw
):
"""
Confirm the sale packing list
"""
order
=
self
.
portal
.
portal_catalog
.
getResultValue
(
uid
=
sequence
[
"sale_packing_list_uid"
])
self
.
assertRaises
(
UnsupportedWorkflowMethod
,
order
.
portal_workflow
.
doActionFor
,
order
,
'confirm_action'
)
def
stepCheckTicAndNotFoundComputerPartitionAvailableCall
(
self
,
sequence
,
**
kw
):
"""
Check that calling ComputerPartition.building raises a NotFoundError
after using tic.
"""
computer_guid
=
sequence
[
"computer_reference"
]
partition_id
=
sequence
[
"computer_partition_reference"
]
self
.
slap
=
slap
.
slap
()
self
.
slap
.
initializeConnection
(
self
.
server_url
,
timeout
=
None
)
computer_partition
=
self
.
slap
.
registerComputerPartition
(
computer_guid
,
partition_id
)
self
.
tic
()
transaction
.
commit
()
self
.
assertRaises
(
slap
.
NotFoundError
,
computer_partition
.
available
)
def
stepCheckTicAndNotFoundComputerPartitionBuildingCall
(
self
,
sequence
,
**
kw
):
"""
Check that calling ComputerPartition.building raises a NotFoundError
after using tic.
"""
computer_guid
=
sequence
[
"computer_reference"
]
partition_id
=
sequence
[
"computer_partition_reference"
]
self
.
slap
=
slap
.
slap
()
self
.
slap
.
initializeConnection
(
self
.
server_url
,
timeout
=
None
)
computer_partition
=
self
.
slap
.
registerComputerPartition
(
computer_guid
,
partition_id
)
self
.
tic
()
transaction
.
commit
()
self
.
assertRaises
(
slap
.
NotFoundError
,
computer_partition
.
building
)
def
stepCheckNotFoundComputerPartitionAvailableAfterRegisterCall
(
self
,
sequence
,
**
kw
):
"""
Check that calling ComputerPartition.available after just registration
raises a NotFoundError
"""
computer_guid
=
sequence
[
"computer_reference"
]
partition_id
=
sequence
[
"computer_partition_reference_list"
][
0
]
self
.
slap
=
slap
.
slap
()
self
.
slap
.
initializeConnection
(
self
.
server_url
,
timeout
=
None
)
computer_partition
=
self
.
slap
.
registerComputerPartition
(
computer_guid
,
partition_id
)
self
.
assertRaises
(
slap
.
NotFoundError
,
computer_partition
.
available
)
def
stepCheckUnauthorizedComputerPartitionAvailableCall
(
self
,
sequence
,
**
kw
):
"""
Check that calling ComputerPartition.available raises an Unauthorized error
"""
computer_guid
=
sequence
[
"computer_reference"
]
partition_id
=
sequence
[
"computer_partition_reference"
]
self
.
slap
=
slap
.
slap
()
self
.
slap
.
initializeConnection
(
self
.
server_url
,
timeout
=
None
)
computer_partition
=
self
.
slap
.
registerComputerPartition
(
computer_guid
,
partition_id
)
self
.
assertRaises
(
slap
.
Unauthorized
,
computer_partition
.
available
)
def
stepCheckNotFoundComputerPartitionAvailableCall
(
self
,
sequence
,
**
kw
):
"""
Check that calling ComputerPartition.available raises a NotFoundError
"""
computer_guid
=
sequence
[
"computer_reference"
]
partition_id
=
sequence
[
"computer_partition_reference"
]
partition_id
=
sequence
[
"computer_partition_reference_list"
][
0
]
self
.
slap
=
slap
.
slap
()
self
.
slap
.
initializeConnection
(
self
.
server_url
,
timeout
=
None
)
computer_partition
=
self
.
slap
.
registerComputerPartition
(
...
...
@@ -2678,37 +2082,6 @@ class TestVifibSlapWebServiceMixin(testVifibMixin):
computer_guid
,
partition_id
)
self
.
assertRaises
(
slap
.
NotFoundError
,
computer_partition
.
destroyed
)
def
stepCheckSuccessComputerPartitionAvailableCall
(
self
,
sequence
,
**
kw
):
"""
Check that calling ComputerPartition.available raises an Unauthorized error
"""
computer_guid
=
sequence
[
"computer_reference"
]
partition_id
=
sequence
[
"computer_partition_reference"
]
self
.
slap
=
slap
.
slap
()
self
.
slap
.
initializeConnection
(
self
.
server_url
,
timeout
=
None
)
computer_partition
=
self
.
slap
.
registerComputerPartition
(
computer_guid
,
partition_id
)
computer_partition
.
available
()
def
stepCheckTicAndUnauthorizedComputerPartitionAvailableCall
(
self
,
sequence
,
**
kw
):
"""
Check that calling ComputerPartition.available raises an Unauthorized error
after using tic.
"""
computer_guid
=
sequence
[
"computer_reference"
]
partition_id
=
sequence
[
"computer_partition_reference"
]
self
.
slap
=
slap
.
slap
()
self
.
slap
.
initializeConnection
(
self
.
server_url
,
timeout
=
None
)
computer_partition
=
self
.
slap
.
registerComputerPartition
(
computer_guid
,
partition_id
)
self
.
tic
()
transaction
.
commit
()
self
.
assertRaises
(
slap
.
Unauthorized
,
computer_partition
.
available
)
def
stepCheckNotFoundComputerPartitionErrorAfterRegisterCall
(
self
,
sequence
,
**
kw
):
"""
...
...
@@ -2722,18 +2095,6 @@ class TestVifibSlapWebServiceMixin(testVifibMixin):
self
.
slap
.
registerComputerPartition
(
computer_guid
,
partition_id
)
def
stepCheckNotFoundComputerPartitionErrorCall
(
self
,
sequence
,
**
kw
):
"""
Check that calling ComputerPartition.error works.
"""
computer_guid
=
sequence
[
"computer_reference"
]
partition_id
=
sequence
[
"computer_partition_reference"
]
self
.
slap
=
slap
.
slap
()
self
.
slap
.
initializeConnection
(
self
.
server_url
,
timeout
=
None
)
computer_partition
=
self
.
slap
.
registerComputerPartition
(
computer_guid
,
partition_id
)
self
.
assertRaises
(
slap
.
NotFoundError
,
computer_partition
.
error
,
"ErrorLog"
)
def
stepCheckSuccessComputerPartitionErrorCall
(
self
,
sequence
,
**
kw
):
"""
Check that calling ComputerPartition.error works.
...
...
@@ -2777,62 +2138,6 @@ class TestVifibSlapWebServiceMixin(testVifibMixin):
uid
=
sequence
[
"sale_packing_list_uid"
])
self
.
assertFalse
(
packing_list
.
getComment
(
''
).
endswith
(
"ErrorLog"
))
def
stepCheckCancelledSalePackingList
(
self
,
sequence
,
**
kw
):
"""
Check that the sale packing list is cancelled
"""
packing_list
=
self
.
portal
.
portal_catalog
.
getResultValue
(
uid
=
sequence
[
"sale_packing_list_uid"
])
self
.
assertEquals
(
"cancelled"
,
packing_list
.
getSimulationState
())
def
stepCheckConfirmedSalePackingList
(
self
,
sequence
,
**
kw
):
"""
Check that the sale packing list is confirmed
"""
packing_list
=
self
.
portal
.
portal_catalog
.
getResultValue
(
uid
=
sequence
[
"sale_packing_list_uid"
])
self
.
assertEquals
(
"confirmed"
,
packing_list
.
getSimulationState
())
def
stepCheckStartedSalePackingList
(
self
,
sequence
,
**
kw
):
"""
Check that the sale packing list is started
"""
packing_list
=
self
.
portal
.
portal_catalog
.
getResultValue
(
uid
=
sequence
[
"sale_packing_list_uid"
])
self
.
assertEquals
(
"started"
,
packing_list
.
getSimulationState
())
def
stepCheckStoppedSalePackingList
(
self
,
sequence
,
**
kw
):
"""
Check that the sale packing list is stopped
"""
packing_list
=
self
.
portal
.
portal_catalog
.
getResultValue
(
uid
=
sequence
[
"sale_packing_list_uid"
])
self
.
assertEquals
(
"stopped"
,
packing_list
.
getSimulationState
())
def
stepCheckDeliveredSalePackingList
(
self
,
sequence
,
**
kw
):
"""
Check that the sale packing list is delivered
"""
packing_list
=
self
.
portal
.
portal_catalog
.
getResultValue
(
uid
=
sequence
[
"sale_packing_list_uid"
])
self
.
assertEquals
(
"delivered"
,
packing_list
.
getSimulationState
())
def
stepCheckTicAndSuccessComputerPartitionErrorCall
(
self
,
sequence
,
**
kw
):
"""
Check that calling ComputerPartition.error works.
"""
computer_guid
=
sequence
[
"computer_reference"
]
partition_id
=
sequence
[
"computer_partition_reference"
]
self
.
slap
=
slap
.
slap
()
self
.
slap
.
initializeConnection
(
self
.
server_url
,
timeout
=
None
)
computer_partition
=
self
.
slap
.
registerComputerPartition
(
computer_guid
,
partition_id
)
self
.
tic
()
transaction
.
commit
()
computer_partition
.
error
(
"ErrorLog"
)
def
stepCheckNotFoundComputerPartitionGetSoftwareReleaseAfterRegisterCall
(
self
,
sequence
,
**
kw
):
"""
...
...
@@ -2865,38 +2170,6 @@ class TestVifibSlapWebServiceMixin(testVifibMixin):
self
.
assertTrue
(
isinstance
(
software_release
,
slap
.
SoftwareRelease
))
self
.
assertEquals
(
url
,
software_release
.
getURI
())
def
stepCheckNotFoundComputerPartitionGetSoftwareReleaseCall
(
self
,
sequence
,
**
kw
):
"""
Check that calling ComputerPartition.getSoftwareRelease raise NotFoundError
"""
computer_guid
=
sequence
[
"computer_reference"
]
partition_id
=
sequence
[
"computer_partition_reference"
]
self
.
slap
=
slap
.
slap
()
self
.
slap
.
initializeConnection
(
self
.
server_url
,
timeout
=
None
)
computer_partition
=
self
.
slap
.
registerComputerPartition
(
computer_guid
,
partition_id
)
self
.
assertRaises
(
slap
.
NotFoundError
,
computer_partition
.
getSoftwareRelease
)
def
stepCheckTicAndNotFoundComputerPartitionGetSoftwareReleaseCall
(
self
,
sequence
,
**
kw
):
"""
Check that calling ComputerPartition.getSoftwareRelease raises an
NotFoundError after using tic.
"""
computer_guid
=
sequence
[
"computer_reference"
]
partition_id
=
sequence
[
"computer_partition_reference"
]
self
.
slap
=
slap
.
slap
()
self
.
slap
.
initializeConnection
(
self
.
server_url
,
timeout
=
None
)
computer_partition
=
self
.
slap
.
registerComputerPartition
(
computer_guid
,
partition_id
)
self
.
tic
()
transaction
.
commit
()
self
.
assertRaises
(
slap
.
NotFoundError
,
computer_partition
.
getSoftwareRelease
)
def
stepCheckSuccessComputerPartitionGetInstanceParameterDictCall
(
self
,
sequence
,
**
kw
):
"""
...
...
@@ -2949,26 +2222,6 @@ class TestVifibSlapWebServiceMixin(testVifibMixin):
}
self
.
assertSameDict
(
expected
,
result
)
def
stepSetSoftwareInstanceValidTestParameterXML
(
self
,
sequence
,
**
kw
):
"""
Set valid XML on the software instance
"""
software_instance_uid
=
sequence
[
"software_instance_uid"
]
software_instance
=
self
.
portal
.
portal_catalog
.
getResultValue
(
uid
=
software_instance_uid
)
test_parameter_id
=
'test_parameter_id'
test_parameter_value
=
'test_parameter_value'
software_instance
.
edit
(
text_content
=
"""<?xml version="1.0" encoding="utf-8"?>
<instance>
<parameter id="%s">%s</parameter>
</instance>"""
%
(
test_parameter_id
,
test_parameter_value
))
sequence
.
edit
(
test_parameter_id
=
test_parameter_id
,
test_parameter_value
=
test_parameter_value
)
def
stepSetSoftwareInstanceValidXML
(
self
,
sequence
,
**
kw
):
"""
Set valid XML on the software instance
...
...
@@ -2998,16 +2251,6 @@ class TestVifibSlapWebServiceMixin(testVifibMixin):
self
.
assertRaises
(
slap
.
NotFoundError
,
requested_slap_computer_partition
.
getInstanceParameterDict
)
def
stepCheckRequestedComputerPartitionTestParameter
(
self
,
sequence
,
**
kw
):
requested_slap_computer_partition
=
sequence
[
'requested_slap_computer_partition'
]
instance_parameter_dict
=
requested_slap_computer_partition
\
.
getInstanceParameterDict
()
test_parameter_id
=
sequence
[
'test_parameter_id'
]
test_parameter_value
=
sequence
[
'test_parameter_value'
]
self
.
assertTrue
(
test_parameter_id
in
instance_parameter_dict
)
self
.
assertEqual
(
test_parameter_value
,
instance_parameter_dict
[
test_parameter_id
])
def
stepCheckRequestedComputerPartitionRequestedParameter
(
self
,
sequence
,
**
kw
):
requested_slap_computer_partition
=
sequence
[
'requested_slap_computer_partition'
]
...
...
@@ -3051,19 +2294,6 @@ class TestVifibSlapWebServiceMixin(testVifibMixin):
}
self
.
assertSameDict
(
expected
,
result
)
def
stepCheckNotFoundComputerPartitionGetStateCall
(
self
,
sequence
,
**
kw
):
"""
Check that ComputerPartition.getState raises a NotFound error
"""
computer_guid
=
sequence
[
"computer_reference"
]
partition_id
=
sequence
[
"computer_partition_reference"
]
self
.
slap
=
slap
.
slap
()
self
.
slap
.
initializeConnection
(
self
.
server_url
,
timeout
=
None
)
computer_partition
=
self
.
slap
.
registerComputerPartition
(
computer_guid
,
partition_id
)
self
.
assertRaises
(
slap
.
NotFoundError
,
computer_partition
.
getState
)
def
stepCheckStoppedComputerPartitionGetStateCall
(
self
,
sequence
,
**
kw
):
"""
Check that ComputerPartition.getState return 'stopped'
...
...
@@ -3104,25 +2334,6 @@ class TestVifibSlapWebServiceMixin(testVifibMixin):
self
.
assertEquals
(
'destroyed'
,
computer_partition
.
getState
())
def
stepCheckTicAndUnauthorizedComputerPartitionGetStateCall
(
self
,
sequence
,
**
kw
):
"""
Check that calling ComputerPartition.getState raises an
Unauthorized error after using tic.
"""
computer_guid
=
sequence
[
"computer_reference"
]
partition_id
=
sequence
[
"computer_partition_reference"
]
self
.
slap
=
slap
.
slap
()
self
.
slap
.
initializeConnection
(
self
.
server_url
,
timeout
=
None
)
computer_partition
=
self
.
slap
.
registerComputerPartition
(
computer_guid
,
partition_id
)
self
.
tic
()
transaction
.
commit
()
self
.
assertRaises
(
slap
.
Unauthorized
,
computer_partition
.
getState
)
def
stepCheckComputerPartitionIsFree
(
self
,
sequence
,
**
kw
):
computer_partition_uid
=
sequence
[
"computer_partition_uid"
]
computer_partition
=
self
.
portal
.
portal_catalog
.
getResultValue
(
...
...
@@ -3669,9 +2880,6 @@ class TestVifibSlapWebServiceMixin(testVifibMixin):
def
stepSetDeliveryLineAmountEqualTwo
(
self
,
sequence
):
sequence
.
edit
(
delivery_line_amount
=
2
)
def
stepSetDeliveryLineAmountEqualThree
(
self
,
sequence
):
sequence
.
edit
(
delivery_line_amount
=
3
)
def
stepSetDeliveryLineAmountEqualOne
(
self
,
sequence
):
sequence
.
edit
(
delivery_line_amount
=
1
)
...
...
master/product/Vifib/tests/testVifibSlapWebServiceSlaveInstance.py
View file @
eb73d8d8
...
...
@@ -478,30 +478,6 @@ class TestVifibSlapWebServiceSlaveInstance(TestVifibSlapWebServiceMixin):
sequence_list
.
addSequenceString
(
sequence_string
)
sequence_list
.
play
(
self
)
def
stepStoreCurrentSoftwareInstanceBufferA
(
self
,
sequence
,
**
kw
):
sequence
.
edit
(
software_instance_uid_buffer_a
=
sequence
[
'software_instance_uid'
],
software_instance_reference_buffer_a
=
sequence
[
'software_instance_reference'
])
def
stepStoreCurrentSoftwareInstanceBufferB
(
self
,
sequence
,
**
kw
):
sequence
.
edit
(
software_instance_uid_buffer_b
=
sequence
[
'software_instance_uid'
],
software_instance_reference_buffer_b
=
sequence
[
'software_instance_reference'
])
def
stepRestoreCurrentSoftwareInstanceBufferA
(
self
,
sequence
,
**
kw
):
sequence
.
edit
(
software_instance_uid
=
sequence
[
'software_instance_uid_buffer_a'
],
software_instance_reference
=
sequence
[
'software_instance_reference_buffer_a'
])
def
stepRestoreCurrentSoftwareInstanceBufferB
(
self
,
sequence
,
**
kw
):
sequence
.
edit
(
software_instance_uid
=
sequence
[
'software_instance_uid_buffer_b'
],
software_instance_reference
=
sequence
[
'software_instance_reference_buffer_b'
])
def
test_SlaveInstance_request_stop
(
self
):
"""
Check that the Slave Instance will be stopped correctly
...
...
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