1- import csv
2- import io
3- from pathlib import Path
4- import re
5- from unittest .mock import patch
61from pathlib import Path
72
83from defusedxml import ElementTree as ET
1510
1611TEST_ASSET_DIR = Path (__file__ ).parent / "assets"
1712
18- BULK_ADD_XML = TEST_ASSET_DIR / "users_bulk_add_job.xml"
1913GET_XML = TEST_ASSET_DIR / "user_get.xml"
2014GET_XML_ALL_FIELDS = TEST_ASSET_DIR / "user_get_all_fields.xml"
2115GET_EMPTY_XML = TEST_ASSET_DIR / "user_get_empty.xml"
3024USERS = TEST_ASSET_DIR / "Data" / "user_details.csv"
3125
3226
33- def make_user (
34- name : str ,
35- site_role : str = "" ,
36- auth_setting : str = "" ,
37- domain : str = "" ,
38- fullname : str = "" ,
39- email : str = "" ,
40- idp_id : str = "" ,
41- ) -> TSC .UserItem :
42- user = TSC .UserItem (name , site_role or None )
43- if auth_setting :
44- user .auth_setting = auth_setting
45- if domain :
46- user ._domain_name = domain
47- if fullname :
48- user .fullname = fullname
49- if email :
50- user .email = email
51- if idp_id :
52- user .idp_configuration_id = idp_id
53- return user
54-
55-
5627@pytest .fixture (scope = "function" )
5728def server ():
5829 """Fixture to create a TSC.Server instance for testing."""
@@ -282,8 +253,7 @@ def test_get_usernames_from_file(server: TSC.Server):
282253 response_xml = ADD_XML .read_text ()
283254 with requests_mock .mock () as m :
284255 m .post (server .users .baseurl , text = response_xml )
285- with pytest .warns (DeprecationWarning ):
286- user_list , failures = server .users .create_from_file (str (USERNAMES ))
256+ user_list , failures = server .users .create_from_file (str (USERNAMES ))
287257 assert user_list [0 ].name == "Cassie" , user_list
288258 assert failures == [], failures
289259
@@ -292,8 +262,7 @@ def test_get_users_from_file(server: TSC.Server):
292262 response_xml = ADD_XML .read_text ()
293263 with requests_mock .mock () as m :
294264 m .post (server .users .baseurl , text = response_xml )
295- with pytest .warns (DeprecationWarning ):
296- users , failures = server .users .create_from_file (str (USERS ))
265+ users , failures = server .users .create_from_file (str (USERS ))
297266 assert users [0 ].name == "Cassie" , users
298267 assert failures == []
299268
@@ -365,209 +334,3 @@ def test_update_user_idp_configuration(server: TSC.Server) -> None:
365334 user_elem = tree .find (".//user" )
366335 assert user_elem is not None
367336 assert user_elem .attrib ["idpConfigurationId" ] == "012345"
368-
369-
370- def test_create_users_csv () -> None :
371- users = [
372- make_user ("Alice" , "Viewer" ),
373- make_user ("Bob" , "Explorer" ),
374- make_user ("Charlie" , "Creator" , "SAML" ),
375- make_user ("Dave" ),
376- make_user ("Eve" , "ServerAdministrator" , "OpenID" , "example.com" , "Eve Example" , "Eve@example.com" ),
377- make_user ("Frank" , "SiteAdministratorExplorer" , "TableauIDWithMFA" , email = "Frank@example.com" ),
378- make_user ("Grace" , "SiteAdministratorCreator" , "SAML" , "example.com" , "Grace Example" , "gex@example.com" ),
379- make_user ("Hank" , "Unlicensed" ),
380- ]
381-
382- license_map = {
383- "Viewer" : "Viewer" ,
384- "Explorer" : "Explorer" ,
385- "ExplorerCanPublish" : "Explorer" ,
386- "Creator" : "Creator" ,
387- "SiteAdministratorExplorer" : "Explorer" ,
388- "SiteAdministratorCreator" : "Creator" ,
389- "ServerAdministrator" : "Creator" ,
390- "Unlicensed" : "Unlicensed" ,
391- }
392- publish_map = {
393- "Unlicensed" : 0 ,
394- "Viewer" : 0 ,
395- "Explorer" : 0 ,
396- "Creator" : 1 ,
397- "ExplorerCanPublish" : 1 ,
398- "SiteAdministratorExplorer" : 1 ,
399- "SiteAdministratorCreator" : 1 ,
400- "ServerAdministrator" : 1 ,
401- }
402- admin_map = {
403- "SiteAdministratorExplorer" : "Site" ,
404- "SiteAdministratorCreator" : "Site" ,
405- "ServerAdministrator" : "System" ,
406- }
407-
408- csv_columns = ["name" , "password" , "fullname" , "license" , "admin" , "publish" , "email" ]
409- csv_data = create_users_csv (users )
410- csv_file = io .StringIO (csv_data .decode ("utf-8" ))
411- csv_reader = csv .reader (csv_file )
412- for user , row in zip (users , csv_reader ):
413- site_role = user .site_role or "Unlicensed"
414- name = f"{ user .domain_name } \\ { user .name } " if user .domain_name else user .name
415- csv_user = dict (zip (csv_columns , row ))
416- assert name == csv_user ["name" ]
417- assert (user .fullname or "" ) == csv_user ["fullname" ]
418- assert (user .email or "" ) == csv_user ["email" ]
419- assert license_map [site_role ] == csv_user ["license" ]
420- assert admin_map .get (site_role , "" ) == csv_user ["admin" ]
421- assert publish_map [site_role ] == int (csv_user ["publish" ])
422-
423-
424- def test_bulk_add (server : TSC .Server ) -> None :
425- server .version = "3.15"
426- users = [
427- make_user ("Alice" , "Viewer" ),
428- make_user ("Bob" , "Explorer" ),
429- make_user ("Charlie" , "Creator" , "SAML" ),
430- make_user ("Dave" ),
431- make_user ("Eve" , "ServerAdministrator" , "OpenID" , "example.com" , "Eve Example" , "Eve@example.com" ),
432- make_user ("Frank" , "SiteAdministratorExplorer" , "TableauIDWithMFA" , email = "Frank@example.com" ),
433- make_user ("Grace" , "SiteAdministratorCreator" , "SAML" , "example.com" , "Grace Example" , "gex@example.com" ),
434- make_user ("Hank" , "Unlicensed" ),
435- make_user ("Ivy" , "Unlicensed" , idp_id = "0123456789" ),
436- ]
437- with requests_mock .mock () as m :
438- m .post (f"{ server .users .baseurl } /import" , text = BULK_ADD_XML .read_text ())
439-
440- job = server .users .bulk_add (users )
441-
442- assert isinstance (job , TSC .JobItem )
443-
444- assert m .last_request .method == "POST"
445- assert m .last_request .url == f"{ server .users .baseurl } /import"
446-
447- body = m .last_request .body .replace (b"\r \n " , b"\n " )
448- assert body .startswith (b"--" ) # Check if it's a multipart request
449- boundary = body .split (b"\n " )[0 ].strip ()
450-
451- # Body starts and ends with a boundary string. Split the body into
452- # segments and ignore the empty sections at the start and end.
453- segments = [seg for s in body .split (boundary ) if (seg := s .strip ()) not in [b"" , b"--" ]]
454- assert len (segments ) == 2 # Check if there are two segments
455-
456- # Check if the first segment is the csv file and the second segment is the xml
457- assert b'Content-Disposition: form-data; name="tableau_user_import"' in segments [0 ]
458- assert b'Content-Disposition: form-data; name="request_payload"' in segments [1 ]
459- assert b"Content-Type: file" in segments [0 ]
460- assert b"Content-Type: text/xml" in segments [1 ]
461-
462- xml_string = segments [1 ].split (b"\n \n " )[1 ].strip ()
463- xml = ET .fromstring (xml_string )
464- xml_users = xml .findall (".//user" , namespaces = {})
465- assert len (xml_users ) == len (users )
466-
467- for user , xml_user in zip (users , xml_users ):
468- assert user .name == xml_user .get ("name" )
469- if user .idp_configuration_id is None :
470- assert xml_user .get ("authSetting" ) == (user .auth_setting or "ServerDefault" )
471- else :
472- assert xml_user .get ("idpConfigurationId" ) == user .idp_configuration_id
473- assert xml_user .get ("authSetting" ) is None
474-
475- csv_data = create_users_csv (users ).replace (b"\r \n " , b"\n " )
476- assert csv_data .strip () == segments [0 ].split (b"\n \n " )[1 ].strip ()
477-
478-
479- def test_bulk_add_no_name (server : TSC .Server ) -> None :
480- server .version = "3.15"
481- users = [
482- TSC .UserItem (site_role = "Viewer" ),
483- ]
484- with requests_mock .mock () as m :
485- m .post (f"{ server .users .baseurl } /import" , text = BULK_ADD_XML .read_text ())
486-
487- with pytest .raises (ValueError , match = "User name must be populated." ):
488- server .users .bulk_add (users )
489-
490-
491- def test_bulk_remove (server : TSC .Server ) -> None :
492- server .version = "3.15"
493- users = [
494- make_user ("Alice" ),
495- make_user ("Bob" , domain = "example.com" ),
496- ]
497- with requests_mock .mock () as m :
498- m .post (f"{ server .users .baseurl } /delete" )
499-
500- server .users .bulk_remove (users )
501-
502- assert m .last_request .method == "POST"
503- assert m .last_request .url == f"{ server .users .baseurl } /delete"
504-
505- body = m .last_request .body .replace (b"\r \n " , b"\n " )
506- assert body .startswith (b"--" ) # Check if it's a multipart request
507- boundary = body .split (b"\n " )[0 ].strip ()
508-
509- content = next (seg for seg in body .split (boundary ) if seg .strip ())
510- assert b'Content-Disposition: form-data; name="tableau_user_delete"' in content
511- assert b"Content-Type: file" in content
512-
513- content = content .replace (b"\r \n " , b"\n " )
514- csv_data = content .split (b"\n \n " )[1 ].decode ("utf-8" )
515- for user , row in zip (users , csv_data .split ("\n " )):
516- name , * _ = row .split ("," )
517- assert name == f"{ user .domain_name } \\ { user .name } " if user .domain_name else user .name
518-
519-
520- def test_add_all (server : TSC .Server ) -> None :
521- server .version = "2.0"
522- users = [
523- make_user ("Alice" , "Viewer" ),
524- make_user ("Bob" , "Explorer" ),
525- make_user ("Charlie" , "Creator" , "SAML" ),
526- make_user ("Dave" ),
527- ]
528-
529- with patch ("tableauserverclient.server.endpoint.users_endpoint.Users.add" , autospec = True ) as mock_add :
530- with pytest .warns (DeprecationWarning ):
531- server .users .add_all (users )
532-
533- assert mock_add .call_count == len (users )
534-
535-
536- def test_add_idp_and_auth_error (server : TSC .Server ) -> None :
537- server .version = "3.24"
538- users = [make_user ("Alice" , "Viewer" , auth_setting = "SAML" , idp_id = "01234" )]
539-
540- with pytest .raises (ValueError , match = "User cannot have both authSetting and idpConfigurationId." ):
541- server .users .bulk_add (users )
542-
543-
544- def test_remove_users_csv (server : TSC .Server ) -> None :
545- server .version = "3.15"
546- users = [
547- make_user ("Alice" , "Viewer" ),
548- make_user ("Bob" , "Explorer" ),
549- make_user ("Charlie" , "Creator" , "SAML" ),
550- make_user ("Dave" ),
551- make_user ("Eve" , "ServerAdministrator" , "OpenID" , "example.com" , "Eve Example" , "Eve@example.com" ),
552- make_user ("Frank" , "SiteAdministratorExplorer" , "TableauIDWithMFA" , email = "Frank@example.com" ),
553- make_user ("Grace" , "SiteAdministratorCreator" , "SAML" , "example.com" , "Grace Example" , "gex@example.com" ),
554- make_user ("Hank" , "Unlicensed" ),
555- make_user ("Ivy" , "Unlicensed" , idp_id = "0123456789" ),
556- ]
557-
558- data = remove_users_csv (users )
559- assert isinstance (data , bytes ), "remove_users_csv should return bytes"
560- csv_data = data .decode ("utf-8" )
561- records = re .split (r"\r?\n" , csv_data .strip ())
562- assert len (records ) == len (users ), "Number of records in csv does not match number of users"
563-
564- for user , record in zip (users , records ):
565- name , * rest = record .strip ().split ("," )
566- assert len (rest ) == 6 , "Number of fields in csv does not match expected number"
567- assert all ([f == "" for f in rest ]), "All fields except name should be empty"
568- if user .domain_name is None :
569- assert name == user .name , f"Name in csv does not match expected name: { user .name } "
570- else :
571- assert (
572- name == f"{ user .domain_name } \\ { user .name } "
573- ), f"Name in csv does not match expected name: { user .domain_name } \\ { user .name } "
0 commit comments