Mod Zeep service to allow basic authentication

primestg/service.py

@@ -1,18 +1,27 @@
 # -*- coding: UTF-8 -*-
 
 from zeep import Client
+from zeep.transports import Transport
+from requests import Session
+from requests.auth import HTTPBasicAuth
 import primestg
 
 
 class Service(object):
-    def __init__(self, fact_id, cnc_url, sync=True, source=None):
-        self.cnc_url = cnc_url
-        self.fact_id = fact_id
-        self.sync = sync
-        if not source:
+    def __init__(self, dc_vals):
+        self.cnc_url = dc_vals['url']
+        self.fact_id = dc_vals['request_id']
+        self.sync = dc_vals['sync']
+        if dc_vals.get('source', False):
+            self.source = dc_vals['source']
+        else:
             self.source = 'DCF'  # By default it doesn't look to the meter for data
+        if dc_vals.get('user', False) and dc_vals.get('password', False):
+            self.auth = True
+            self.user = dc_vals['user']
+            self.password = dc_vals['password']
         else:
-            self.source = source
+            self.auth = False
         self.DC_service = self.create_service()
 
     def send(self, report_id, meters, date_from='', date_to=''):
@@ -29,7 +38,13 @@ def send(self, report_id, meters, date_from='', date_to=''):
 
     def create_service(self):
         binding = '{http://www.asais.fr/ns/Saturne/DC/ws}WS_DCSoap'
-        client = Client(wsdl=primestg.get_data('WS_DC.wsdl'))
+        if self.auth:
+            session = Session()
+            session.auth = HTTPBasicAuth(self.user, self.password)
+            client = Client(wsdl=primestg.get_data('WS_DC.wsdl'),
+                            transport=Transport(session=session))
+        else:
+            client = Client(wsdl=primestg.get_data('WS_DC.wsdl'))
         client.set_ns_prefix(None, 'http://www.asais.fr/ns/Saturne/DC/ws')
         return client.create_service(binding, self.cnc_url)
 

spec/Report_S02_spec.py

@@ -62,7 +62,5 @@
             with open(result_filename) as result_file:
                 result_string = result_file.read()
                 expected_result = literal_eval(result_string)
-
-        result = self.report[key].values
-
-        expect(result).to(equal(expected_result))
+                result = self.report[key].values
+                expect(result).to(equal(expected_result))

spec/web_service_spec.py

@@ -4,7 +4,13 @@
 
 with description('Web services run'):
     with before.all:
-        self.s = Service(1, 'http://cct.gisce.lan:8080/')
+        vals = {
+            'url': 'http://cct.gisce.lan:8080/',
+            'request_id': 12,
+            'sync': False,
+            'source': 'DCF',
+        }
+        self.s = Service(vals)
 
     with it('asking for S02 report with mocked connection'):
         with responses.RequestsMock() as rsps: