Added/Updated functional\services\test_user_management.py: checked on 4.0.1.2692, 3.0.8.33535

tests/functional/services/test_user_management.py

@@ -21,79 +21,74 @@ FBTEST:      functional.services.user_management
 
 import pytest
 from firebird.qa import *
+import firebird.driver
+from firebird.driver import TPB, Isolation, core as fb_core
 
-db = db_factory()
+db = db_factory(init = "create sequence g;")
 
 act = python_act('db')
 
-expected_stdout = """
-    Adding user.
-    Done.
-    Search in users list.
-    Found user: TMP$TEST$USER$
-    Trying to establish connection.
-    TMP$TEST$USER$
-    TCP
-    Done.
-    Removing user.
-    Done.
+test_expected_stdout = """
+    POINT:                           1
+    SEC$USER_NAME:                   TMP$TEST$USER
+    SEC$FIRST_NAME:                  John
+    SEC$LAST_NAME:                   Smith
+    SEC$ADMIN:                       True
+
+    POINT:                           2
+    SEC$USER_NAME:                   TMP$TEST$USER
+    SEC$FIRST_NAME:                  Robert
+    SEC$LAST_NAME:                   Jackson
+    SEC$ADMIN:                       False
+
+    POINT:                           3
+    SEC$USER_NAME:                   None
+    SEC$FIRST_NAME:                  None
+    SEC$LAST_NAME:                   None
+    SEC$ADMIN:                       None
 """
 
-@pytest.mark.skip('FIXME: Not IMPLEMENTED')
 @pytest.mark.version('>=3')
-def test_1(act: Action):
-    pytest.fail("Not IMPLEMENTED")
+def test_1(act: Action, capsys):
+    
+    #----------------------------------------------------------
+    def print_user_data(con, cur, prepared_sttm):
+        con.commit()
+        con.begin()
+        cur.execute(prepared_sttm, (TMP_USER_NAME,))
+        col_hdr=cur.description
+        for r in cur:
+            for i in range(0,len(col_hdr)):
+                print( (col_hdr[i][0] +':').ljust(32), r[i] )
+    #----------------------------------------------------------
 
-# test_script_1
-#---
-#
-#  import os
-#  import subprocess
-#  from fdb import services
-#
-#  TEST_USER_NAME='tmp$test$user$'.upper()
-#  db_conn.close()
-#
-#  svc_con = services.connect( host='localhost', user = user_name, password = user_password )
-#
-#  u01 = services.User(TEST_USER_NAME)
-#  u01.password = 'QweRty'
-#  u01.first_name = 'Foo'
-#  u01.last_name = 'Bar'
-#
-#
-#  print('Adding user.')
-#  svc_con.add_user(u01)
-#  print('Done.')
-#
-#  usr_list = svc_con.get_users()
-#  print('Search in users list.')
-#  for u in usr_list:
-#      if u.name == TEST_USER_NAME:
-#          print('Found user:', u.name)
-#
-#  sql='''
-#      select
-#           mon$user as user_connected
-#          ,iif( upper(mon$remote_protocol) starting with upper('TCP'), 'TCP', coalesce(mon$remote_protocol, '<NULL>') ) as protocol_info
-#      from mon$attachments
-#      where mon$attachment_id = current_connection
-#  '''
-#
-#  try:
-#      print('Trying to establish connection.')
-#      usr_con = fdb.connect( dsn = dsn, user=TEST_USER_NAME, password='QweRty')
-#      cur2 = usr_con.cursor()
-#      cur2.execute( sql )
-#      for r in cur2:
-#          for i in range(0,len(r)):
-#              print( ''.join( ( r[i] ) ) )
-#      print('Done.')
-#  finally:
-#      usr_con.close()
-#
-#
-#  print('Removing user.')
-#  svc_con.remove_user(u01)
-#  print('Done.')
-#---
+    TMP_USER_NAME = 'tmp$test$user'.upper()
+    sttm = 'select gen_id(g,1) as point, sec$user_name, sec$first_name, sec$last_name, sec$admin from rdb$database left join sec$users on sec$user_name = ?'
+
+    with act.db.connect() as con:
+        con.begin()
+        cur = con.cursor()
+        prepared_sttm = cur.prepare(sttm)
+
+        with act.connect_server() as srv:
+            svc = fb_core.ServerUserServices(srv)
+            if svc.exists(user_name = TMP_USER_NAME):
+                svc.delete(user_name = TMP_USER_NAME)
+
+            svc.add( user_name = TMP_USER_NAME, password = '123', first_name = 'John', last_name = 'Smith', admin = True)
+            print_user_data(con, cur, prepared_sttm)
+
+            # Here we make sure that user actually exists and can make connecttion:
+            with act.db.connect(user = TMP_USER_NAME, password = '123') as con_check:
+                pass
+
+            svc.update( user_name = TMP_USER_NAME, last_name = 'Jackson', admin = False, first_name = 'Robert')
+            print_user_data(con, cur, prepared_sttm)
+
+            svc.delete(user_name = TMP_USER_NAME)
+            print_user_data(con, cur, prepared_sttm)
+
+    act.expected_stdout = test_expected_stdout
+    act.stdout = capsys.readouterr().out
+
+    assert act.clean_stdout == act.clean_expected_stdout