aboutsummaryrefslogtreecommitdiffstats
diff options
context:
space:
mode:
-rw-r--r--gsxws/comms.py42
-rw-r--r--tests/test_gsxws.py32
2 files changed, 69 insertions, 5 deletions
diff --git a/gsxws/comms.py b/gsxws/comms.py
index c2896fb..dc37e9b 100644
--- a/gsxws/comms.py
+++ b/gsxws/comms.py
@@ -4,14 +4,52 @@ from core import GsxObject
class Communication(GsxObject):
- def get_content():
+
+ _namespace = "glob:"
+
+ def get_content(self):
"""
The Fetch Communication Content API allows the service providers/depot/carriers
to fetch the communication content by article ID from the service news channel.
"""
+ return self._submit("lookupRequestData", "FetchCommunicationContent",
+ "communicationMessage")
- def get_articles():
+ def get_articles(self):
"""
The Fetch Communication Articles API allows the service partners
to fetch all the active communication message IDs.
"""
+ return self._submit("lookupRequestData", "FetchCommunicationArticles",
+ "communicationMessage")
+
+ def acknowledge(self):
+ """
+ The Acknowledge Communication API allows the service providers/depot/carriers to
+ update the status as Read/UnRead.
+ """
+ return self._submit("communicationRequest", "AcknowledgeCommunication",
+ "communicationResponse")
+
+
+def fetch(**kwargs):
+ return Communication(**kwargs).get_articles()
+
+
+def content(id):
+ return Communication(articleID=id).get_content()
+
+
+def ack(id, status):
+ ack = GsxObject(articleID=id)
+ ack.acknowledgeType = status
+ return Communication(acknowledgement=ack).acknowledge()
+
+
+if __name__ == '__main__':
+ import sys
+ import doctest
+ from core import connect
+ logging.basicConfig(level=logging.DEBUG)
+ connect(*sys.argv[1:4])
+ doctest.testmod()
diff --git a/tests/test_gsxws.py b/tests/test_gsxws.py
index c4924da..164c90c 100644
--- a/tests/test_gsxws.py
+++ b/tests/test_gsxws.py
@@ -10,13 +10,39 @@ from gsxws.core import validate
from gsxws.objectify import parse, gsx_diags_timestamp
from gsxws.products import Product
from gsxws import (repairs, escalations, lookups, returns,
- GsxError, ServicePart, diagnostics, comptia,)
+ GsxError, ServicePart, diagnostics, comptia,
+ comms,)
def empty(a):
return a in [None, '', ' ']
+class CommsTestCase(TestCase):
+ def setUp(self):
+ from gsxws.core import connect
+ self.priority = 'HIGH'
+ self.article_id = 'SN3133'
+ connect(os.getenv('GSX_USER'), os.getenv('GSX_SOLDTO'), os.getenv('GSX_ENV'))
+ self.articles = comms.fetch(priority=self.priority, readStatus=False)
+
+ def test_priority(self):
+ for a in self.articles:
+ self.assertEqual(a.priority, self.priority)
+
+ def test_date(self):
+ for a in self.articles:
+ self.assertIsInstance(a.createdDate, date)
+
+ def test_content(self):
+ content = comms.content(self.article_id)
+ self.assertEqual(content.languageCode, 'en')
+
+ def test_ack(self):
+ result = comms.ack(self.article_id, 'UNREAD')
+ self.assertEqual(result.acknowledgeType, 'UNREAD')
+
+
class RemoteTestCase(TestCase):
def setUp(self):
from gsxws.core import connect
@@ -177,9 +203,9 @@ class TestErrorFunctions(TestCase):
from gsxws.core import GsxResponse
xml = open('tests/fixtures/error_ca_fmip.xml', 'r').read()
with self.assertRaisesRegexp(GsxError, 'A repair cannot be created'):
- GsxResponse(xml=xml, el_method='CreateCarryInResponse',
+ GsxResponse(xml=xml, el_method='CreateCarryInResponse',
el_response='repairConfirmation')
-
+
class TestLookupFunctions(RemoteTestCase):
def test_component_check(self):