Commit 66f8be9b authored by Rafael Monnerat's avatar Rafael Monnerat

slapos_rss: Fix RSS Scripts and add unit tests

Several aspects were broken and not working.
Include Unit tests for the scripts as it was missing.
parent 97651fa6
""" """
Keep a custom script for permit render other times of documents, ie.: Software Installation. Keep a custom script for permit render other times of documents, ie.: Software Installation.
""" """
from Products.ZSQLCatalog.SQLCatalog import ComplexQuery, SimpleQuery
portal = context.getPortalObject() portal = context.getPortalObject()
kw['portal_type'] = ["Support Request", "Upgrade Decision"]
# Source project is only set for Support Request, it don't work for query = ComplexQuery(
# Upgrade Decisions. ComplexQuery(
kw['source_project_uid'] = context.getUid() SimpleQuery(portal_type="Support Request"),
SimpleQuery(default_source_project_uid=context.getUid()),
logical_operator='and'),
ComplexQuery(
SimpleQuery(portal_type="Upgrade Decision Line"),
SimpleQuery(default_aggregate_uid=context.getUid()),
logical_operator='and'),
logical_operator='or')
person = portal.ERP5Site_getAuthenticatedMemberPersonValue() # Use event modification date instead.
if person: kw['sort_on'] = [('modification_date', 'DESC'),]
kw['default_destination_decision_uid'] = person.getUid() kw['simulation_state'] = "NOT cancelled"
result_list = []
# Use event modification date instead. for document in portal.portal_catalog(query=query, **kw):
kw['sort_on'] = [('modification_date', 'DESC'),] if document.getPortalType() == "Upgrade Decision Line":
return context.getPortalObject().portal_catalog(**kw) result_list.append(document.getParentValue())
continue
else: result_list.append(document)
return [] return result_list
""" """
Keep a custom script for permit render other times of documents, ie.: Software Installation. Keep a custom script for permit render other times of documents, ie.: Software Installation.
""" """
from Products.ZSQLCatalog.SQLCatalog import ComplexQuery, SimpleQuery
portal = context.getPortalObject() portal = context.getPortalObject()
kw['portal_type'] = ["Support Request", "Upgrade Decision"]
kw['source_project_uid'] = context.getUid()
person = portal.ERP5Site_getAuthenticatedMemberPersonValue() query = ComplexQuery(
if person: ComplexQuery(
kw['default_destination_decision_uid'] = person.getUid() SimpleQuery(portal_type="Support Request"),
SimpleQuery(default_source_project_uid=context.getUid()),
# Use event modification date instead. logical_operator='and'),
kw['sort_on'] = [('modification_date', 'DESC'),] ComplexQuery(
return context.getPortalObject().portal_catalog(**kw) SimpleQuery(portal_type="Upgrade Decision Line"),
SimpleQuery(default_aggregate_uid=context.getUid()),
logical_operator='and'),
logical_operator='or')
else: # Use event modification date instead.
return [] kw['sort_on'] = [('modification_date', 'DESC'),]
kw['simulation_state'] = "NOT cancelled"
result_list = []
for document in portal.portal_catalog(query=query, **kw):
if document.getPortalType() == "Upgrade Decision Line":
result_list.append(document.getParentValue())
continue
result_list.append(document)
return result_list
...@@ -2,10 +2,11 @@ portal = context.getPortalObject() ...@@ -2,10 +2,11 @@ portal = context.getPortalObject()
event = portal.portal_catalog.getResultValue( event = portal.portal_catalog.getResultValue(
follow_up_uid=context.getUid(), follow_up_uid=context.getUid(),
sort_on=[('delivery.start_date', 'DESC')], sort_on=[('modification_date', 'DESC')],
) )
if event: if event:
return event.getModificationDate() return event.getModificationDate()
else:
context.getModificationDate() return context.getModificationDate()
...@@ -2,10 +2,10 @@ portal = context.getPortalObject() ...@@ -2,10 +2,10 @@ portal = context.getPortalObject()
event = portal.portal_catalog.getResultValue( event = portal.portal_catalog.getResultValue(
follow_up_uid=context.getUid(), follow_up_uid=context.getUid(),
sort_on=[('delivery.start_date', 'DESC')], sort_on=[('modification_date', 'DESC')],
) )
if event: if event:
return event.getTextContent() return event.getTextContent()
else:
context.getDescription() return context.getDescription()
portal = context.getPortalObject() portal = context.getPortalObject()
kw['portal_type'] = ["Support Request", "Regularisation Request", "Upgrade Decision"] kw['portal_type'] = ["Support Request", "Regularisation Request", "Upgrade Decision"]
person = portal.ERP5Site_getAuthenticatedMemberPersonValue() person = portal.ERP5Site_getAuthenticatedMemberPersonValue()
if person: if person:
kw['simulation_state'] = "NOT cancelled"
kw['default_destination_decision_uid'] = person.getUid() kw['default_destination_decision_uid'] = person.getUid()
kw['sort_on'] = [('modification_date', 'DESC'),] kw['sort_on'] = [('modification_date', 'DESC'),]
return context.getPortalObject().portal_catalog(**kw) return context.getPortalObject().portal_catalog( **kw)
else: else:
return [] return []
##############################################################################
#
# Copyright (c) 2002-2016 Nexedi SA and Contributors. All Rights Reserved.
#
# WARNING: This program as such is intended to be used by professional
# programmers who take the whole responsibility of assessing all potential
# consequences resulting from its eventual inadequacies and bugs
# End users who are looking for a ready-to-use solution with commercial
# guarantees and support are strongly adviced to contract a Free Software
# Service Company
#
# This program is Free Software; you can redistribute it and/or
# modify it under the terms of the GNU General Public License
# as published by the Free Software Foundation; either version 2
# of the License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
#
##############################################################################
# Copyright (c) 2013 Nexedi SA and Contributors. All Rights Reserved.
import transaction
from Products.SlapOS.tests.testSlapOSMixin import \
testSlapOSMixin
from Products.ERP5Type.tests.SecurityTestCase import SecurityTestCase
import time
from DateTime import DateTime
class TestSlapOSPerson_checkToCreateRegularisationRequest(testSlapOSMixin,
SecurityTestCase):
#def beforeTearDown(self):
# transaction.abort()
def createPerson(self):
person_user = self.portal.person_module.template_member.\
Base_createCloneDocument(batch_mode=1)
person_user.edit(
title="live_test_%s" % self.new_id,
reference="live_test_%s" % self.new_id,
default_email_text="live_test_%s@example.org" % self.new_id,
)
person_user.validate()
for assignment in person_user.contentValues(portal_type="Assignment"):
assignment.open()
person_user.immediateReindexObject()
transaction.commit()
return person_user
def createComputer(self, person=None):
computer = self.portal.computer_module.newContent(
portal_type="Computer",
title="%s Test RSS Computer" % self.new_id)
computer.edit(source_administration_value=person)
computer.validate()
computer.immediateReindexObject()
computer.updateLocalRolesOnSecurityGroups()
return computer
def createHostingSubscription(self, person=None):
hs = self.portal.hosting_subscription_module.newContent(
portal_type="Hosting Subscription",
title="%s Test RSS Hosting Subscription" % self.new_id)
if person is not None:
hs.edit(destination_section=person.getRelativeUrl())
hs.validate()
hs.immediateReindexObject()
hs.updateLocalRolesOnSecurityGroups()
return hs
def createSupportRequest(self, person, computer=None, validate=1):
sr = self.portal.support_request_module.newContent(\
title="Test Support Request %s" % self.new_id,
resource="service_module/slapos_crm_monitoring",
destination_decision_value=person)
if validate:
sr.validate()
if computer:
sr.setSourceProjectValue(computer)
sr.immediateReindexObject()
sr.updateLocalRolesOnSecurityGroups()
if validate:
self.assertUserCanAccessDocument(person.getReference(), sr)
self.assertUserCanViewDocument(person.getReference(), sr)
return sr
def createUpgradeDecision(self, person, computer=None, validate=1):
ud = self.portal.upgrade_decision_module.newContent(\
title="Upgrade Decision %s" % self.new_id,
destination_decision_value=person)
ud.immediateReindexObject()
ud.updateLocalRolesOnSecurityGroups()
# This seems odd, user can see draft SR even he is auditor
self.assertUserCanAccessDocument(person.getReference(), ud)
self.assertUserCanViewDocument(person.getReference(), ud)
if computer is not None:
line = ud.newContent(
portal_type="Upgrade Decision Line",
aggregate_value=computer)
line.immediateReindexObject()
return ud
def createRegularisationRequest(self, person, validate=1):
rr = self.portal.regularisation_request_module.newContent(
portal_type='Regularisation Request',
title="Test Reg. Req.%s" % self.new_id,
reference="TESTREGREQ-%s" % self.new_id,
destination_decision_value=person)
if validate:
rr.validate()
rr.immediateReindexObject()
rr.updateLocalRolesOnSecurityGroups()
if validate:
self.assertUserCanAccessDocument(person.getReference(), rr)
self.assertUserCanViewDocument(person.getReference(), rr)
return rr
def afterSetUp(self):
""" bootstrap test """
self.login()
self.changeSkin("View")
if getattr(self, "person", None) is None:
self.new_id = self.generateNewId()
self.person = self.createPerson()
self.computer = self.createComputer(person=self.person)
self.hosting_subscription = self.createHostingSubscription(
person=self.person)
self.support_request = self.createSupportRequest(self.person,
computer=self.computer)
self.upgrade_decision = self.createUpgradeDecision(self.person,
computer=self.computer)
self.hs_support_request = self.createSupportRequest(self.person,
computer=self.hosting_subscription)
self.hs_upgrade_decision = self.createUpgradeDecision(self.person,
computer=self.hosting_subscription)
self.regularisation_request = self.createRegularisationRequest(
self.person)
self.cancelled_support_request = self.createSupportRequest(self.person,
computer=self.computer,
validate=0)
self.cancelled_upgrade_decision = self.createUpgradeDecision(self.person,
computer=self.computer,
validate=0)
self.cancelled_regularisation_request = self.createRegularisationRequest(
self.person, validate=0)
self.cancelled_support_request.cancel()
self.cancelled_upgrade_decision.cancel()
self.cancelled_regularisation_request.cancel()
self.tic()
self.login(self.person.getReference())
self.changeSkin("RSS")
def test_WebSection_getUserRssTicketList(self):
""" Test get User RSS """
ticket_list = self.portal.WebSection_getUserRssTicketList()
self.assertEquals(len(ticket_list), 5)
self.assertSameSet([i.getUid() for i in ticket_list],
[self.support_request.getUid(),
self.upgrade_decision.getUid(),
self.regularisation_request.getUid(),
self.hs_support_request.getUid(),
self.hs_upgrade_decision.getUid()])
def _test_WebSection_getRSSContent(self, ticket):
""" Test for get Date and Content for the RSS Feed
"""
date_content = ticket.WebSection_getRSSDateContent()
description = ticket.WebSection_getRSSDescriptionContent()
self.assertEquals(ticket.getModificationDate(), date_content)
self.assertEquals(ticket.getDescription(), description)
event = self.portal.event_module.newContent(
portal_type="Web Message",
follow_up_value=ticket,
title="%s Web Message" % ticket.getTitle(),
text_content="%s Web Message Content" % ticket.getTitle(),
start_date=DateTime()
)
event.immediateReindexObject()
transaction.commit()
# Now the event dates and content should be showed instead of the ticket
self.assertEquals(event.getModificationDate(),
ticket.WebSection_getRSSDateContent())
self.assertEquals(event.getTextContent(),
ticket.WebSection_getRSSDescriptionContent())
# Create another event and check if it continues the same
new_event = self.portal.event_module.newContent(
portal_type="Web Message",
follow_up_value=ticket,
title="%s New Web Message" % ticket.getTitle(),
text_content="%s New Web Message Content" % ticket.getTitle(),
start_date=DateTime()
)
new_event.immediateReindexObject()
transaction.commit()
# Now the new event dates and content should be showed instead
self.assertEquals(new_event.getModificationDate(),
ticket.WebSection_getRSSDateContent())
self.assertEquals(new_event.getTextContent(),
ticket.WebSection_getRSSDescriptionContent())
# Catalog table dont index microseconds, so it can cause problems when search
# sleep 2 second to produce a good modification date.
time.sleep(2)
# Modify first event title and check if it takes the lead again.
event.setTitle("%s Modified Web Message" % ticket.getTitle())
event.edit(start_date=DateTime())
event.immediateReindexObject()
transaction.commit()
self.assertTrue(event.getModificationDate() > new_event.getModificationDate(),
"%s is not larger them %s" % (event.getModificationDate(), new_event.getModificationDate()))
# Now the first event dates and content should be showed instead
self.assertEquals(event.getTextContent(),
ticket.WebSection_getRSSDescriptionContent())
self.assertEquals(event.getModificationDate(),
ticket.WebSection_getRSSDateContent())
def test_WebSection_getRSSContents_support_request(self):
""" Test for get Date and Content for the RSS Feed for
Support Request
"""
self._test_WebSection_getRSSContent(self.support_request)
def test_WebSection_getRSSContents_upgrade_decision(self):
""" Test for get Date and Content for the RSS Feed for
Upgrade Decision
"""
self._test_WebSection_getRSSContent(self.upgrade_decision)
def test_WebSection_getRSSContents_regularisation_request(self):
""" Test for get Date and Content for the RSS Feed for
Upgrade Decision
"""
self._test_WebSection_getRSSContent(self.regularisation_request)
def test_Computer_getUserRssEntryList(self):
""" Test get Computerr RSS """
self.login()
ticket_list = self.computer.Computer_getUserRssEntryList()
self.assertSameSet([i.getRelativeUrl() for i in ticket_list],
[self.support_request.getRelativeUrl(),
self.upgrade_decision.getRelativeUrl()])
self.assertEquals(len(ticket_list), 2)
def test_HostingSubscription_getUserRssEntryList(self):
""" Test get Hosting Subscription RSS """
self.login()
ticket_list = self.hosting_subscription.HostingSubscription_getUserRssEntryList()
self.assertSameSet([i.getRelativeUrl() for i in ticket_list],
[self.hs_support_request.getRelativeUrl(),
self.hs_upgrade_decision.getRelativeUrl()])
self.assertEquals(len(ticket_list), 2)
<?xml version="1.0"?>
<ZopeData>
<record id="1" aka="AAAAAAAAAAE=">
<pickle>
<global name="Test Component" module="erp5.portal_type"/>
</pickle>
<pickle>
<dictionary>
<item>
<key> <string>_recorded_property_dict</string> </key>
<value>
<persistent> <string encoding="base64">AAAAAAAAAAI=</string> </persistent>
</value>
</item>
<item>
<key> <string>default_reference</string> </key>
<value> <string>testSlapOSRSSSkins</string> </value>
</item>
<item>
<key> <string>description</string> </key>
<value>
<none/>
</value>
</item>
<item>
<key> <string>id</string> </key>
<value> <string>test.erp5.testSlapOSRSSSkins</string> </value>
</item>
<item>
<key> <string>portal_type</string> </key>
<value> <string>Test Component</string> </value>
</item>
<item>
<key> <string>sid</string> </key>
<value>
<none/>
</value>
</item>
<item>
<key> <string>text_content_error_message</string> </key>
<value>
<tuple/>
</value>
</item>
<item>
<key> <string>text_content_warning_message</string> </key>
<value>
<tuple/>
</value>
</item>
<item>
<key> <string>version</string> </key>
<value> <string>erp5</string> </value>
</item>
<item>
<key> <string>workflow_history</string> </key>
<value>
<persistent> <string encoding="base64">AAAAAAAAAAM=</string> </persistent>
</value>
</item>
</dictionary>
</pickle>
</record>
<record id="2" aka="AAAAAAAAAAI=">
<pickle>
<global name="PersistentMapping" module="Persistence.mapping"/>
</pickle>
<pickle>
<dictionary>
<item>
<key> <string>data</string> </key>
<value>
<dictionary/>
</value>
</item>
</dictionary>
</pickle>
</record>
<record id="3" aka="AAAAAAAAAAM=">
<pickle>
<global name="PersistentMapping" module="Persistence.mapping"/>
</pickle>
<pickle>
<dictionary>
<item>
<key> <string>data</string> </key>
<value>
<dictionary>
<item>
<key> <string>component_validation_workflow</string> </key>
<value>
<persistent> <string encoding="base64">AAAAAAAAAAQ=</string> </persistent>
</value>
</item>
</dictionary>
</value>
</item>
</dictionary>
</pickle>
</record>
<record id="4" aka="AAAAAAAAAAQ=">
<pickle>
<global name="WorkflowHistoryList" module="Products.ERP5Type.patches.WorkflowTool"/>
</pickle>
<pickle>
<tuple>
<none/>
<list>
<dictionary>
<item>
<key> <string>action</string> </key>
<value> <string>validate</string> </value>
</item>
<item>
<key> <string>validation_state</string> </key>
<value> <string>validated</string> </value>
</item>
</dictionary>
</list>
</tuple>
</pickle>
</record>
</ZopeData>
test.erp5.testSlapOSRSSSkins
\ No newline at end of file
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment