add simple (dummy) test for external (zitadel) user creation

scopes/org/user.py

@@ -20,7 +20,7 @@ class User:
     firstName: str = ''
     lastName: str = ''
     displayName: str = ''
-    grants: List[str] = field(default_factory=list)
+    groups: List[str] = field(default_factory=list)
 
     def __post_init__(self):
         if not self.displayName:
@@ -61,7 +61,7 @@ class ExtUser:
         )
         return data
 
-    def create(self, updateIfExits=False):
+    def create(self, updateIfExists=False):
         data = self.asDict()
         if self.user.hashedPassword:
             data['hashedPassword'] = self.user.hashedPassword
@@ -69,10 +69,9 @@ class ExtUser:
         if status > 201:
             if updateIfExits:
                 return self.update()
-            else:
         return status, res
-        if self.user.grants:
-            return self.createGrants()
+        #if self.user.groups:
+            #return self.createGroups()
 
     def update(self, createIfMissing=False):
         data = self.asDict()
@@ -84,19 +83,13 @@ class ExtUser:
                 return self.create()
             else:
                 return status, res
-        if self.user.grants:
-            #return self.updateGrants()
-            groups = ' '.join(self.user.grants)
-            data = dict(metadata=[dict(key='gloops', value='groups')])
-            return self.client.post(f'v2/users/human/{self.userId}/metadata', data)
+        #if self.user.groups:
+            #return self.updateGroups()
 
-    def createGrants(self):
+    def createGroups(self):
         data = dict(
                 userId=self.userId,
                 projectId=config.oidc_params['project_id'],
-                roleKeys=self.user.grants,
+                roleKeys=self.user.groups,
         )
         return self.client.post(self.endpoints['create_authorization'], data)
-
-    def updateGrants(self):
-        self.createGrants()

scopes/tests/config.py

@@ -54,7 +54,10 @@ oidc_params = dict(
     cookie_domain=getenv('OIDC_COOKIE_DOMAIN', None),
     cookie_lifetime=getenv('OIDC_COOKIE_LIFETIME', '86400'),
     cookie_crypt=getenv('OIDC_COOKIE_CRYPT', None),
-    private_key_file=getenv('OIDC_SERVICE_USER_PRIVATE_KEY_FILE', '.private-key.json'),
+    private_key_file=getenv('OIDC_SERVICE_USER_PRIVATE_KEY_FILE', 
+                            'scopes/tests/test-private-key.json'),
+    organization_id=getenv('OIDC_ORGANIZATION_ID', '12346'),
+    project_id=getenv('OIDC_PROJECT_ID', '12347'),
 )
 
 oidc_provider_endpoints = dict(

scopes/tests/data_auth.py

@@ -26,7 +26,12 @@ oidc_data = {
         "kid": "316766976250797901",
         "alg": "RS256",
         "n": public_key_n,
-        "e": "AQAB"}]}
+        "e": "AQAB"}]},
+    'test://oidc/oauth/v2/token': {
+        "access_token": "abcde12345"},
+    'test://oidc/v2/users/human': {
+        "code": 1}
+    }
 }
 
 from scopes.tests.dummy_requests import response_data

scopes/tests/dummy_requests.py

@@ -9,12 +9,20 @@ def get(url, *args, **kw):
     logger.info(f'get: %s - %s - %s', url, args, kw)
     return FakeResponse(response_data[url])
 
+def post(url, *args, **kw):
+    logger.info(f'post: %s - %s - %s', url, args, kw)
+    return FakeResponse(response_data[url])
+
 
 class FakeResponse:
 
     def __init__(self, data):
         self.data = data
 
+    @property
+    def status_code(self):
+        return 200
+
     def json(self):
         return self.data
 

scopes/tests/test-private-key.json

@@ -0,0 +1 @@
+{"type":"serviceaccount","keyId":"314794985486606157","key":"-----BEGIN RSA PRIVATE KEY-----\nMIIEowIBAAKCAQEA0dC8wcwu6Uefxx/shqsSTk//ATseeCy28RMAEa4NFGj/y8Ju\nOfVUj7pB5+6onjmsBAHXhCJ+fEWWAJdHnbvywrBNNhVx38f8v+90zUP2IzlT1UDp\ncTIYaehnf3+uqwgMcijnYJ6UgaHFMSecxnYD4adnw8J/FEMDgy2N+v5krp989VQ5\nT2kgrkb/l5z8dgLhmmcLKm7YCG1uXXP+g+qzEZ9Uhur5b+czjIalzC/tq2V2JoJB\nooH9w1iaRXRKel7FZPo0YGyQh/0a9Zn5JsXVc3YTHTKh9madr/yQqmk+6siTl/Ou\ntz9mvpY+AfFRaIWikoyB3W9rHd0b6WtQPflEPwIDAQABAoIBAAN64daZC2IlJPpJ\nhkPJjJkt7H3ZvCykGTiwZvzkFSV0hGGdzPQ7JHbp0PQG2lcdf8PlP+zaIZzwDofd\n+nscRe+CuxUdj/D1QTTxxM8uxGNbLQ/JbtXIzezbxPOxa3U8wfAWy5enqbDovPuO\nu6PzCydv/mGZ1T/ByMohNEyocYUP6mupHWwf2hN/lnrL264w8uvNjAw0xDtbtBJN\nX61u6vi/fiY37qKblN3irAePwK4LIhHZZoyJ1HrFYIkFf0Bviuzpw/ASVqbjizPV\nmTxGxghiQacAMvSSe+pcfJ7ip74rCFv7+6pzL+yW8df1lbSM9vS+86SDgY9RCc2E\n3h1/hUECgYEA/WqiWNXey25qCNB6WHo3SU5cZIZVNWzsT1zkwkXOUtEyU0/zEfT+\nEjW/vbxIBgZNV1tX2aXd7Ke5OCoQ1dqLnmDoO5d13xTeaWN3FR8ibTwbaDCwyg5d\njyIXK2k7IwtcpJFgJFGM/6udAdO/bPm1IPEslJXHBqZoGrKb+bTw6N8CgYEA0/RQ\nHtQluQYBtXNzEql0MaxBUxfHkwjL6Yo6dM+EJAomI+cccVy22s+z2aQX5GVQnbzs\nm9BGkJzzn7eGPy3i2LgStqUZ2W7VqfIJNCIDbC7OxBAaszh5/LEgv5pfp1Yr/HIf\nwHZz53rdV8H+oUfMJdlyrRyGOeGIDZCd94nTMKECgYAQOpT9BW1IL+EAgYFkSydh\nPXBzS5sHWdtkVbmcq2XELfuAFF2np73hoqmN2BHwuNSZJJNir9mffzpAW4lKeL16\nPhCBSHjW+Xoo26LTqnPE9RV4Pa4EspjRQsijEhEkdGTRcTHsAYD7Gp1qcYoPy4oK\n+wb02Qau6Vc/ZnLQsgK/lwKBgQDMLSGxUPQ11E95GAnWBF7mKuWSwemC/opQItRF\nClJk1VIAa/W+Tm3nQwYhti0920tZaFEVmAEh9c/KH+S2n+FSm5+LSmgoSNiSqZGs\nIsfhQwXzYQAXfWQlxAukB3X1oNEmkll78Z+dcYIfs8UyYBOMsngBwuSahWOmjZVe\ni+phgQKBgC0ozpbIcNg48M4/Rrev3qJB7XlU74MySsFJdBhlrzmK3+z02bXWbyaJ\nzQLwC6Dorw0PcWAKtcJcbBn6ZAoptcmG6wdQrYk1IC+82TDcNvAFL06y8OXHYLtu\ni5AiE4nK1waoDF/1I66VACyKI6hhISRW3bKaxHhrx5OsGKVurF4R\n-----END RSA PRIVATE KEY-----\n","expirationDate":"9999-12-31T23:59:59Z","userId":"311889729668833101"}

scopes/tests/tlib_web.py

@@ -43,3 +43,7 @@ def test_auth(self, config):
 
 def test_user_data(self, config):
     from scopes.org import user
+    u1 = user.User(login='tst9', email='tst9@example.com',
+                   firstName='Test', lastName='User')
+    xu1 = user.ExtUser(u1, idPrefix='test.')
+    xu1.create(updateIfExists=True)