feat: test of DS api added

Author: ymarcon

Makefile

@@ -4,6 +4,12 @@ install:
 test:
 	poetry run pytest
 
+test-api-admin:
+	poetry run pytest tests/test_api_admin.py
+
+test-api-analysis:
+	poetry run pytest tests/test_api_analysis.py
+
 build:
 	poetry build
 

datashield_opal/impl.py

@@ -4,7 +4,7 @@
 
 from argparse import Namespace
 from obiba_opal.core import OpalClient, UriBuilder, OpalRequest, OpalResponse, HTTPError
-from datashield.interface import DSDriver, DSConnection, DSResult, DSError
+from datashield.interface import DSLoginInfo, DSDriver, DSConnection, DSResult, DSError
 
 
 class OpalDSError(DSError):
@@ -291,12 +291,10 @@ def _delete(self, ws) -> OpalRequest:
 class OpalDriver(DSDriver):
 
     @classmethod
-    def new_connection(cls, name: str, args: dict, profile: str = 'default', restore: str = None) -> DSConnection:
-        namedArgs = Namespace(opal = args['url'], user = args['user'], password = args['password'], token = args['token'])
+    def new_connection(cls, args: DSLoginInfo, restore: str = None) -> DSConnection:
+        namedArgs = Namespace(opal = args.url, user = args.user, password = args.password, token = args.token)
         loginInfo = OpalClient.LoginInfo.parse(namedArgs)
-        return OpalConnection(name, loginInfo, profile, restore)
-
-
+        return OpalConnection(args.name, loginInfo, args.profile, restore)
 
 class OpalResult(DSResult):
 

tests/test_api_admin.py

@@ -0,0 +1,48 @@
+from datashield import DSSession, DSLoginBuilder
+
+class TestClass:
+
+    @classmethod
+    def setup_class(cls):
+        #url = 'http://localhost:8080'
+        url = 'https://opal-demo.obiba.org'
+        builder = DSLoginBuilder().add('server1', url, 'dsuser', 'P@ssw0rd').add('server2', url, 'dsuser', 'P@ssw0rd')
+        logins = builder.build()
+        session = DSSession(logins)
+        session.open()
+        setattr(cls, 'session', session)
+
+    @classmethod
+    def teardown_class(cls):
+        cls.session.close()
+    
+    def test_profiles(self):
+        rval = self.session.profiles()
+        assert 'server1' in rval
+        assert rval['server1']['current'] == 'default'
+        assert len(rval['server1']['available']) > 0
+
+    def test_packages(self):
+        rval = self.session.packages()
+        print(rval)
+        assert True
+    
+    def test_methods(self):
+        rval = self.session.methods()
+        print(rval)
+        assert True
+
+    def test_tables(self):
+        rval = self.session.tables()
+        print(rval)
+        assert True
+
+    def test_resources(self):
+        rval = self.session.resources()
+        print(rval)
+        assert True
+
+    def test_workspaces(self):
+        rval = self.session.workspaces()
+        print(rval)
+        assert True

tests/test_api_analysis.py

@@ -0,0 +1,59 @@
+from datashield import DSSession, DSLoginBuilder
+
+class TestClass:
+
+    def setup_method(self, method):
+        #url = 'http://localhost:8080'
+        url = 'https://opal-demo.obiba.org'
+        builder = DSLoginBuilder().add('server1', url, 'dsuser', 'P@ssw0rd').add('server2', url, 'dsuser', 'P@ssw0rd')
+        logins = builder.build()
+        self.session = DSSession(logins)
+        self.session.open()
+
+    def teardown_method(self, method):
+        self.session.close()
+    
+    def test_assign_tables(self):
+        self.session.assign_table('df', tables = { 'server1': 'CNSIM.CNSIM1', 'server2': 'CNSIM.CNSIM2' }, asynchronous = False)
+        rval = self.session.ls()
+        assert len(rval) == 2
+        assert 'server1' in rval
+        assert rval['server1'] == ['df']
+        assert 'server2' in rval
+        assert rval['server2'] == ['df']
+    
+    def test_assign_table(self):
+        self.session.assign_table('df', table = 'CNSIM.CNSIM1', asynchronous = False)
+        rval = self.session.ls()
+        assert len(rval) == 2
+        assert 'server1' in rval
+        assert rval['server1'] == ['df']
+        assert 'server2' in rval
+        assert rval['server2'] == ['df']
+
+    def test_assign_resources(self):
+        self.session.assign_resource('client', resources = { 'server1': 'RSRC.CNSIM1', 'server2': 'RSRC.CNSIM2' }, asynchronous = False)
+        rval = self.session.ls()
+        assert len(rval) == 2
+        assert 'server1' in rval
+        assert rval['server1'] == ['client']
+        assert 'server2' in rval
+        assert rval['server2'] == ['client']
+
+        self.session.assign_expr('df', 'as.resource.data.frame(client, strict = TRUE)', asynchronous = False)
+        rval = self.session.ls()
+        assert len(rval) == 2
+        assert 'server1' in rval
+        assert len(rval['server1']) == 2
+        assert 'client' in rval['server1']
+        assert 'df' in rval['server1']
+    
+    def test_assign_resource(self):
+        self.session.assign_resource('client', resource = 'RSRC.CNSIM1', asynchronous = False)
+        rval = self.session.ls()
+        assert len(rval) == 2
+        assert 'server1' in rval
+        assert rval['server1'] == ['client']
+        assert 'server2' in rval
+        assert rval['server2'] == ['client']
+        

tests/test_datashield.py tests/test_impl.pytests/test_datashield.py renamed to tests/test_impl.py

@@ -1,23 +1,25 @@
-from datashield.interface import DSDriver, DSError
-from datashield import DSLoginBuilder
+from datashield import DSError, DSLoginBuilder, DSSession
 import time
 
 
 class TestClass:
 
     @classmethod
     def setup_class(cls):
-        driver = DSDriver.load_class('datashield_opal.OpalDriver')
         #url = 'http://localhost:8080'
         url = 'https://opal-demo.obiba.org'
         builder = DSLoginBuilder().add('server1', url, 'dsuser', 'P@ssw0rd')
         logins = builder.build()
-        conn = driver.new_connection(logins[0]['name'], logins[0], profile = logins[0]['profile'])
-        setattr(cls, 'conn', conn)
+        session = DSSession(logins)
+        session.open()
+        print(session.has_errors())
+        print(session.has_connections())
+        setattr(cls, 'session', session)
+        setattr(cls, 'conn', session.conns[0])
 
     @classmethod
     def teardown_class(cls):
-        cls.conn.disconnect()    
+        cls.session.close()
 
     def test_driver(self):
         conn = self.conn