c4267cf4c40fb1f866b5958ff122ef836b8c5dfbTimo Sirainenimport os

c4267cf4c40fb1f866b5958ff122ef836b8c5dfbTimo Sirainenimport stat

c4267cf4c40fb1f866b5958ff122ef836b8c5dfbTimo Sirainenimport ent

fdc557286bc9f92c5f3bb49096ff6e2bcec0ea79Timo Sirainenimport grp

c4267cf4c40fb1f866b5958ff122ef836b8c5dfbTimo Sirainenimport pwd

c4267cf4c40fb1f866b5958ff122ef836b8c5dfbTimo Sirainenimport config

c4267cf4c40fb1f866b5958ff122ef836b8c5dfbTimo Sirainenimport signal

c4267cf4c40fb1f866b5958ff122ef836b8c5dfbTimo Sirainenimport subprocess

c4267cf4c40fb1f866b5958ff122ef836b8c5dfbTimo Sirainenimport time

c4267cf4c40fb1f866b5958ff122ef836b8c5dfbTimo Sirainenimport pytest

150e64c376365becf1ec5c9d45912ecb840eea96Timo Sirainenimport ds_openldap

150e64c376365becf1ec5c9d45912ecb840eea96Timo Sirainenimport ldap_ent

c4267cf4c40fb1f866b5958ff122ef836b8c5dfbTimo Sirainenimport sssd_id

c4267cf4c40fb1f866b5958ff122ef836b8c5dfbTimo Sirainenfrom util import unindent

c4267cf4c40fb1f866b5958ff122ef836b8c5dfbTimo Sirainen

c4267cf4c40fb1f866b5958ff122ef836b8c5dfbTimo SirainenLDAP_BASE_DN = "dc=example,dc=com"

39e6fcc3e8b1ccb13087c232cb6bdea04d1a20a4Timo Sirainen

c4267cf4c40fb1f866b5958ff122ef836b8c5dfbTimo Sirainen

c4267cf4c40fb1f866b5958ff122ef836b8c5dfbTimo Sirainen@pytest.fixture(scope="module")

bc3698b8892df8003b410daea6f5bbcd20433808Timo Sirainendef ds_inst(request):

c4267cf4c40fb1f866b5958ff122ef836b8c5dfbTimo Sirainen """LDAP server instance fixture"""

c4267cf4c40fb1f866b5958ff122ef836b8c5dfbTimo Sirainen ds_inst = ds_openldap.DSOpenLDAP(

c4267cf4c40fb1f866b5958ff122ef836b8c5dfbTimo Sirainen config.PREFIX, 10389, LDAP_BASE_DN,

c4267cf4c40fb1f866b5958ff122ef836b8c5dfbTimo Sirainen "cn=admin", "Secret123")

c4267cf4c40fb1f866b5958ff122ef836b8c5dfbTimo Sirainen try:

c4267cf4c40fb1f866b5958ff122ef836b8c5dfbTimo Sirainen ds_inst.setup()

39e6fcc3e8b1ccb13087c232cb6bdea04d1a20a4Timo Sirainen except:

c4267cf4c40fb1f866b5958ff122ef836b8c5dfbTimo Sirainen ds_inst.teardown()

c4267cf4c40fb1f866b5958ff122ef836b8c5dfbTimo Sirainen raise

fdc557286bc9f92c5f3bb49096ff6e2bcec0ea79Timo Sirainen request.addfinalizer(lambda: ds_inst.teardown())

fdc557286bc9f92c5f3bb49096ff6e2bcec0ea79Timo Sirainen return ds_inst

c4267cf4c40fb1f866b5958ff122ef836b8c5dfbTimo Sirainen

bc3698b8892df8003b410daea6f5bbcd20433808Timo Sirainen

bc3698b8892df8003b410daea6f5bbcd20433808Timo Sirainen@pytest.fixture(scope="module")

88187ee880b4829443e0d55ea7d145d9d5880217Timo Sirainendef ldap_conn(request, ds_inst):

88187ee880b4829443e0d55ea7d145d9d5880217Timo Sirainen """LDAP server connection fixture"""

88187ee880b4829443e0d55ea7d145d9d5880217Timo Sirainen ldap_conn = ds_inst.bind()

c4267cf4c40fb1f866b5958ff122ef836b8c5dfbTimo Sirainen ldap_conn.ds_inst = ds_inst

c4267cf4c40fb1f866b5958ff122ef836b8c5dfbTimo Sirainen request.addfinalizer(lambda: ldap_conn.unbind_s())

c4267cf4c40fb1f866b5958ff122ef836b8c5dfbTimo Sirainen return ldap_conn

c4267cf4c40fb1f866b5958ff122ef836b8c5dfbTimo Sirainen

d756ebcfa96bd7cff02097c8f26df9df368b81b1Timo Sirainen

d756ebcfa96bd7cff02097c8f26df9df368b81b1Timo Sirainendef create_ldap_fixture(request, ldap_conn, ent_list):

ccb77e2f63626ec46e5745ef4f38baa8e8e504fcTimo Sirainen """Add LDAP entries and add teardown for removing them"""

c4267cf4c40fb1f866b5958ff122ef836b8c5dfbTimo Sirainen for entry in ent_list:

d5cebe7f98e63d4e2822863ef2faa4971e8b3a5dTimo Sirainen ldap_conn.add_s(entry[0], entry[1])

fdc557286bc9f92c5f3bb49096ff6e2bcec0ea79Timo Sirainen

bd74402ca1a39ec303075fefb1212d7e18a71531Timo Sirainen def teardown():

5fb3bff645380804c9db2510940c41db6b8fdb01Timo Sirainen for entry in ent_list:

5fb3bff645380804c9db2510940c41db6b8fdb01Timo Sirainen ldap_conn.delete_s(entry[0])

d5cebe7f98e63d4e2822863ef2faa4971e8b3a5dTimo Sirainen request.addfinalizer(teardown)

c4267cf4c40fb1f866b5958ff122ef836b8c5dfbTimo Sirainen

c4267cf4c40fb1f866b5958ff122ef836b8c5dfbTimo Sirainen

c4267cf4c40fb1f866b5958ff122ef836b8c5dfbTimo Sirainendef create_conf_fixture(request, contents):

c4267cf4c40fb1f866b5958ff122ef836b8c5dfbTimo Sirainen """Generate sssd.conf and add teardown for removing it"""

ccb77e2f63626ec46e5745ef4f38baa8e8e504fcTimo Sirainen conf = open(config.CONF_PATH, "w")

c4267cf4c40fb1f866b5958ff122ef836b8c5dfbTimo Sirainen conf.write(contents)

c4267cf4c40fb1f866b5958ff122ef836b8c5dfbTimo Sirainen conf.close()

c4267cf4c40fb1f866b5958ff122ef836b8c5dfbTimo Sirainen os.chmod(config.CONF_PATH, stat.S_IRUSR | stat.S_IWUSR)

bd74402ca1a39ec303075fefb1212d7e18a71531Timo Sirainen request.addfinalizer(lambda: os.unlink(config.CONF_PATH))

bd74402ca1a39ec303075fefb1212d7e18a71531Timo Sirainen

bd74402ca1a39ec303075fefb1212d7e18a71531Timo Sirainen

bd74402ca1a39ec303075fefb1212d7e18a71531Timo Sirainendef stop_sssd():

bd74402ca1a39ec303075fefb1212d7e18a71531Timo Sirainen pid_file = open(config.PIDFILE_PATH, "r")

ef11d3930c3602fc86349a4e3a53442df470b601Timo Sirainen pid = int(pid_file.read())

c6ae908f6a2313573625d782bdd4e0ff3882c44aTimo Sirainen os.kill(pid, signal.SIGTERM)

c6ae908f6a2313573625d782bdd4e0ff3882c44aTimo Sirainen while True:

150e64c376365becf1ec5c9d45912ecb840eea96Timo Sirainen try:

ef11d3930c3602fc86349a4e3a53442df470b601Timo Sirainen os.kill(pid, signal.SIGCONT)

ef11d3930c3602fc86349a4e3a53442df470b601Timo Sirainen except:

ef11d3930c3602fc86349a4e3a53442df470b601Timo Sirainen break

ef11d3930c3602fc86349a4e3a53442df470b601Timo Sirainen time.sleep(1)

ef11d3930c3602fc86349a4e3a53442df470b601Timo Sirainen

ef11d3930c3602fc86349a4e3a53442df470b601Timo Sirainen

ef11d3930c3602fc86349a4e3a53442df470b601Timo Sirainendef create_sssd_fixture(request):

ef11d3930c3602fc86349a4e3a53442df470b601Timo Sirainen """Start sssd and add teardown for stopping it and removing state"""

ef11d3930c3602fc86349a4e3a53442df470b601Timo Sirainen if subprocess.call(["sssd", "-D", "-f"]) != 0:

c4267cf4c40fb1f866b5958ff122ef836b8c5dfbTimo Sirainen raise Exception("sssd start failed")

c4267cf4c40fb1f866b5958ff122ef836b8c5dfbTimo Sirainen

c4267cf4c40fb1f866b5958ff122ef836b8c5dfbTimo Sirainen def teardown():

c4267cf4c40fb1f866b5958ff122ef836b8c5dfbTimo Sirainen try:

ccb77e2f63626ec46e5745ef4f38baa8e8e504fcTimo Sirainen stop_sssd()

c4267cf4c40fb1f866b5958ff122ef836b8c5dfbTimo Sirainen except:

c4267cf4c40fb1f866b5958ff122ef836b8c5dfbTimo Sirainen pass

c4267cf4c40fb1f866b5958ff122ef836b8c5dfbTimo Sirainen subprocess.call(["sss_cache", "-E"])

c4267cf4c40fb1f866b5958ff122ef836b8c5dfbTimo Sirainen for path in os.listdir(config.DB_PATH):

16c89b1260c9d07c01c83a9219424d3727069b2eTimo Sirainen os.unlink(config.DB_PATH + "/" + path)

16c89b1260c9d07c01c83a9219424d3727069b2eTimo Sirainen for path in os.listdir(config.MCACHE_PATH):

d756ebcfa96bd7cff02097c8f26df9df368b81b1Timo Sirainen os.unlink(config.MCACHE_PATH + "/" + path)

68efcccb384f2d6871164b072457e87473502c51Timo Sirainen request.addfinalizer(teardown)

d756ebcfa96bd7cff02097c8f26df9df368b81b1Timo Sirainen

d756ebcfa96bd7cff02097c8f26df9df368b81b1Timo Sirainen

d756ebcfa96bd7cff02097c8f26df9df368b81b1Timo Sirainendef load_data_to_ldap(request, ldap_conn):

d756ebcfa96bd7cff02097c8f26df9df368b81b1Timo Sirainen ent_list = ldap_ent.List(LDAP_BASE_DN)

c4267cf4c40fb1f866b5958ff122ef836b8c5dfbTimo Sirainen ent_list.add_user("user1", 1001, 2001)

c4267cf4c40fb1f866b5958ff122ef836b8c5dfbTimo Sirainen ent_list.add_user("user2", 1002, 2002)

c4267cf4c40fb1f866b5958ff122ef836b8c5dfbTimo Sirainen ent_list.add_user("user3", 1003, 2003)

c979eeda1f46483d9c963e265786b701d7683d77Timo Sirainen ent_list.add_user("user11", 1011, 2001)

c979eeda1f46483d9c963e265786b701d7683d77Timo Sirainen ent_list.add_user("user12", 1012, 2002)

c979eeda1f46483d9c963e265786b701d7683d77Timo Sirainen ent_list.add_user("user13", 1013, 2003)

c4267cf4c40fb1f866b5958ff122ef836b8c5dfbTimo Sirainen ent_list.add_user("user21", 1021, 2001)

68efcccb384f2d6871164b072457e87473502c51Timo Sirainen ent_list.add_user("user22", 1022, 2002)

c4267cf4c40fb1f866b5958ff122ef836b8c5dfbTimo Sirainen ent_list.add_user("user23", 1023, 2003)

c4267cf4c40fb1f866b5958ff122ef836b8c5dfbTimo Sirainen

a54be2bd26d6f0860d194d3aeedfa6b7fc14d24cTimo Sirainen ent_list.add_group("group1", 2001, ["user1", "user11", "user21"])

a54be2bd26d6f0860d194d3aeedfa6b7fc14d24cTimo Sirainen ent_list.add_group("group2", 2002, ["user2", "user12", "user22"])

a54be2bd26d6f0860d194d3aeedfa6b7fc14d24cTimo Sirainen ent_list.add_group("group3", 2003, ["user3", "user13", "user23"])

10cfe8a2bdc5ccfc05380689c71c27209327538fTimo Sirainen

10cfe8a2bdc5ccfc05380689c71c27209327538fTimo Sirainen ent_list.add_group("group0x", 2000, ["user1", "user2", "user3"])

a54be2bd26d6f0860d194d3aeedfa6b7fc14d24cTimo Sirainen ent_list.add_group("group1x", 2010, ["user11", "user12", "user13"])

10cfe8a2bdc5ccfc05380689c71c27209327538fTimo Sirainen ent_list.add_group("group2x", 2020, ["user21", "user22", "user23"])

a54be2bd26d6f0860d194d3aeedfa6b7fc14d24cTimo Sirainen create_ldap_fixture(request, ldap_conn, ent_list)

a54be2bd26d6f0860d194d3aeedfa6b7fc14d24cTimo Sirainen

a54be2bd26d6f0860d194d3aeedfa6b7fc14d24cTimo Sirainen

c4267cf4c40fb1f866b5958ff122ef836b8c5dfbTimo Sirainen@pytest.fixture

d756ebcfa96bd7cff02097c8f26df9df368b81b1Timo Sirainendef sanity_rfc2307(request, ldap_conn):

16c89b1260c9d07c01c83a9219424d3727069b2eTimo Sirainen load_data_to_ldap(request, ldap_conn)

c4267cf4c40fb1f866b5958ff122ef836b8c5dfbTimo Sirainen

c4267cf4c40fb1f866b5958ff122ef836b8c5dfbTimo Sirainen conf = unindent("""\

39e6fcc3e8b1ccb13087c232cb6bdea04d1a20a4Timo Sirainen """).format(**locals())

39e6fcc3e8b1ccb13087c232cb6bdea04d1a20a4Timo Sirainen create_conf_fixture(request, conf)

16c89b1260c9d07c01c83a9219424d3727069b2eTimo Sirainen create_sssd_fixture(request)

d756ebcfa96bd7cff02097c8f26df9df368b81b1Timo Sirainen return None

88286b0527bcc0711e312e9db65ca121a45213e3Timo Sirainen

68efcccb384f2d6871164b072457e87473502c51Timo Sirainen

d756ebcfa96bd7cff02097c8f26df9df368b81b1Timo Sirainen@pytest.fixture

d756ebcfa96bd7cff02097c8f26df9df368b81b1Timo Sirainendef fqname_rfc2307(request, ldap_conn):

d756ebcfa96bd7cff02097c8f26df9df368b81b1Timo Sirainen load_data_to_ldap(request, ldap_conn)

d42c9a8f362b76740418c4f9f9441eb7fc661e57Timo Sirainen

c4267cf4c40fb1f866b5958ff122ef836b8c5dfbTimo Sirainen conf = unindent("""\

d42c9a8f362b76740418c4f9f9441eb7fc661e57Timo Sirainen """).format(**locals())

d756ebcfa96bd7cff02097c8f26df9df368b81b1Timo Sirainen create_conf_fixture(request, conf)

d756ebcfa96bd7cff02097c8f26df9df368b81b1Timo Sirainen create_sssd_fixture(request)

88286b0527bcc0711e312e9db65ca121a45213e3Timo Sirainen return None

88286b0527bcc0711e312e9db65ca121a45213e3Timo Sirainen

88286b0527bcc0711e312e9db65ca121a45213e3Timo Sirainen

88286b0527bcc0711e312e9db65ca121a45213e3Timo Sirainen@pytest.fixture

88286b0527bcc0711e312e9db65ca121a45213e3Timo Sirainendef fqname_case_insensitive_rfc2307(request, ldap_conn):

c4267cf4c40fb1f866b5958ff122ef836b8c5dfbTimo Sirainen load_data_to_ldap(request, ldap_conn)

d756ebcfa96bd7cff02097c8f26df9df368b81b1Timo Sirainen

c4267cf4c40fb1f866b5958ff122ef836b8c5dfbTimo Sirainen conf = unindent("""\

88286b0527bcc0711e312e9db65ca121a45213e3Timo Sirainen """).format(**locals())

c4267cf4c40fb1f866b5958ff122ef836b8c5dfbTimo Sirainen create_conf_fixture(request, conf)

42507d758b053bb483de58fba55c73a9eb5d3fbaTimo Sirainen create_sssd_fixture(request)

42507d758b053bb483de58fba55c73a9eb5d3fbaTimo Sirainen return None

42507d758b053bb483de58fba55c73a9eb5d3fbaTimo Sirainen

42507d758b053bb483de58fba55c73a9eb5d3fbaTimo Sirainen

42507d758b053bb483de58fba55c73a9eb5d3fbaTimo Sirainendef test_getpwnam(ldap_conn, sanity_rfc2307):

42507d758b053bb483de58fba55c73a9eb5d3fbaTimo Sirainen ent.assert_passwd_by_name(

d756ebcfa96bd7cff02097c8f26df9df368b81b1Timo Sirainen 'user1',

88286b0527bcc0711e312e9db65ca121a45213e3Timo Sirainen dict(name='user1', passwd='*', uid=1001, gid=2001,

88286b0527bcc0711e312e9db65ca121a45213e3Timo Sirainen gecos='1001', shell='/bin/bash'))

88286b0527bcc0711e312e9db65ca121a45213e3Timo Sirainen ent.assert_passwd_by_uid(

edf8c2b4ef6be8af262bc09aac751eae669670caTimo Sirainen 1001,

c4267cf4c40fb1f866b5958ff122ef836b8c5dfbTimo Sirainen dict(name='user1', passwd='*', uid=1001, gid=2001,

e60a349c641bb2f4723e4a395a25f55531682d2bTimo Sirainen gecos='1001', shell='/bin/bash'))

e60a349c641bb2f4723e4a395a25f55531682d2bTimo Sirainen

d756ebcfa96bd7cff02097c8f26df9df368b81b1Timo Sirainen ent.assert_passwd_by_name(

c4267cf4c40fb1f866b5958ff122ef836b8c5dfbTimo Sirainen 'user2',

42507d758b053bb483de58fba55c73a9eb5d3fbaTimo Sirainen dict(name='user2', passwd='*', uid=1002, gid=2002,

42507d758b053bb483de58fba55c73a9eb5d3fbaTimo Sirainen gecos='1002', shell='/bin/bash'))

c4267cf4c40fb1f866b5958ff122ef836b8c5dfbTimo Sirainen ent.assert_passwd_by_uid(

c4267cf4c40fb1f866b5958ff122ef836b8c5dfbTimo Sirainen 1002,

c4267cf4c40fb1f866b5958ff122ef836b8c5dfbTimo Sirainen dict(name='user2', passwd='*', uid=1002, gid=2002,

6ef7e31619edfaa17ed044b45861d106a86191efTimo Sirainen gecos='1002', shell='/bin/bash'))

6ef7e31619edfaa17ed044b45861d106a86191efTimo Sirainen

c4267cf4c40fb1f866b5958ff122ef836b8c5dfbTimo Sirainen ent.assert_passwd_by_name(

42507d758b053bb483de58fba55c73a9eb5d3fbaTimo Sirainen 'user3',

42507d758b053bb483de58fba55c73a9eb5d3fbaTimo Sirainen dict(name='user3', passwd='*', uid=1003, gid=2003,

150e64c376365becf1ec5c9d45912ecb840eea96Timo Sirainen gecos='1003', shell='/bin/bash'))

42507d758b053bb483de58fba55c73a9eb5d3fbaTimo Sirainen ent.assert_passwd_by_uid(

42507d758b053bb483de58fba55c73a9eb5d3fbaTimo Sirainen 1003,

c4267cf4c40fb1f866b5958ff122ef836b8c5dfbTimo Sirainen dict(name='user3', passwd='*', uid=1003, gid=2003,

42507d758b053bb483de58fba55c73a9eb5d3fbaTimo Sirainen gecos='1003', shell='/bin/bash'))

42507d758b053bb483de58fba55c73a9eb5d3fbaTimo Sirainen

e60a349c641bb2f4723e4a395a25f55531682d2bTimo Sirainen ent.assert_passwd_by_name(

c4267cf4c40fb1f866b5958ff122ef836b8c5dfbTimo Sirainen 'user11',

d756ebcfa96bd7cff02097c8f26df9df368b81b1Timo Sirainen dict(name='user11', passwd='*', uid=1011, gid=2001,

d756ebcfa96bd7cff02097c8f26df9df368b81b1Timo Sirainen gecos='1011', shell='/bin/bash'))

150e64c376365becf1ec5c9d45912ecb840eea96Timo Sirainen ent.assert_passwd_by_uid(

150e64c376365becf1ec5c9d45912ecb840eea96Timo Sirainen 1011,

150e64c376365becf1ec5c9d45912ecb840eea96Timo Sirainen dict(name='user11', passwd='*', uid=1011, gid=2001,

150e64c376365becf1ec5c9d45912ecb840eea96Timo Sirainen gecos='1011', shell='/bin/bash'))

150e64c376365becf1ec5c9d45912ecb840eea96Timo Sirainen

150e64c376365becf1ec5c9d45912ecb840eea96Timo Sirainen ent.assert_passwd_by_name(

18634dae6e304bac982bb1e0ff1b6b88fc448dbcTimo Sirainen 'user12',

294f1a51763015cda0e2d874c5027d6fe7a2cd54Timo Sirainen dict(name='user12', passwd='*', uid=1012, gid=2002,

e60a349c641bb2f4723e4a395a25f55531682d2bTimo Sirainen gecos='1012', shell='/bin/bash'))

e60a349c641bb2f4723e4a395a25f55531682d2bTimo Sirainen ent.assert_passwd_by_uid(

42507d758b053bb483de58fba55c73a9eb5d3fbaTimo Sirainen 1012,

39e6fcc3e8b1ccb13087c232cb6bdea04d1a20a4Timo Sirainen dict(name='user12', passwd='*', uid=1012, gid=2002,

150e64c376365becf1ec5c9d45912ecb840eea96Timo Sirainen gecos='1012', shell='/bin/bash'))

d756ebcfa96bd7cff02097c8f26df9df368b81b1Timo Sirainen

e60a349c641bb2f4723e4a395a25f55531682d2bTimo Sirainen ent.assert_passwd_by_name(

e60a349c641bb2f4723e4a395a25f55531682d2bTimo Sirainen 'user13',

42507d758b053bb483de58fba55c73a9eb5d3fbaTimo Sirainen dict(name='user13', passwd='*', uid=1013, gid=2003,

42507d758b053bb483de58fba55c73a9eb5d3fbaTimo Sirainen gecos='1013', shell='/bin/bash'))

42507d758b053bb483de58fba55c73a9eb5d3fbaTimo Sirainen ent.assert_passwd_by_uid(

42507d758b053bb483de58fba55c73a9eb5d3fbaTimo Sirainen 1013,

edf8c2b4ef6be8af262bc09aac751eae669670caTimo Sirainen dict(name='user13', passwd='*', uid=1013, gid=2003,

e60a349c641bb2f4723e4a395a25f55531682d2bTimo Sirainen gecos='1013', shell='/bin/bash'))

d756ebcfa96bd7cff02097c8f26df9df368b81b1Timo Sirainen

ent.assert_passwd_by_name(

'user21',

dict(name='user21', passwd='*', uid=1021, gid=2001,

gecos='1021', shell='/bin/bash'))

ent.assert_passwd_by_uid(

1021,

dict(name='user21', passwd='*', uid=1021, gid=2001,

gecos='1021', shell='/bin/bash'))

ent.assert_passwd_by_name(

'user22',

dict(name='user22', passwd='*', uid=1022, gid=2002,

gecos='1022', shell='/bin/bash'))

ent.assert_passwd_by_uid(

1022,

dict(name='user22', passwd='*', uid=1022, gid=2002,

gecos='1022', shell='/bin/bash'))

ent.assert_passwd_by_name(

'user23',

dict(name='user23', passwd='*', uid=1023, gid=2003,

gecos='1023', shell='/bin/bash'))

ent.assert_passwd_by_uid(

1023,

dict(name='user23', passwd='*', uid=1023, gid=2003,

gecos='1023', shell='/bin/bash'))

def test_getpwnam_with_mc(ldap_conn, sanity_rfc2307):

test_getpwnam(ldap_conn, sanity_rfc2307)

stop_sssd()

test_getpwnam(ldap_conn, sanity_rfc2307)

def test_getgrnam_simple(ldap_conn, sanity_rfc2307):

ent.assert_group_by_name("group1", dict(name="group1", gid=2001))

ent.assert_group_by_gid(2001, dict(name="group1", gid=2001))

ent.assert_group_by_name("group2", dict(name="group2", gid=2002))

ent.assert_group_by_gid(2002, dict(name="group2", gid=2002))

ent.assert_group_by_name("group3", dict(name="group3", gid=2003))

ent.assert_group_by_gid(2003, dict(name="group3", gid=2003))

ent.assert_group_by_name("group0x", dict(name="group0x", gid=2000))

ent.assert_group_by_gid(2000, dict(name="group0x", gid=2000))

ent.assert_group_by_name("group1x", dict(name="group1x", gid=2010))

ent.assert_group_by_gid(2010, dict(name="group1x", gid=2010))

ent.assert_group_by_name("group2x", dict(name="group2x", gid=2020))

ent.assert_group_by_gid(2020, dict(name="group2x", gid=2020))

def test_getgrnam_simple_with_mc(ldap_conn, sanity_rfc2307):

test_getgrnam_simple(ldap_conn, sanity_rfc2307)

stop_sssd()

test_getgrnam_simple(ldap_conn, sanity_rfc2307)

def test_getgrnam_membership(ldap_conn, sanity_rfc2307):

ent.assert_group_by_name(

"group1",

dict(mem=ent.contains_only("user1", "user11", "user21")))

ent.assert_group_by_gid(

2001,

dict(mem=ent.contains_only("user1", "user11", "user21")))

ent.assert_group_by_name(

"group2",

dict(mem=ent.contains_only("user2", "user12", "user22")))

ent.assert_group_by_gid(

2002,

dict(mem=ent.contains_only("user2", "user12", "user22")))

ent.assert_group_by_name(

"group3",

dict(mem=ent.contains_only("user3", "user13", "user23")))

ent.assert_group_by_gid(

2003,

dict(mem=ent.contains_only("user3", "user13", "user23")))

ent.assert_group_by_name(

"group0x",

dict(mem=ent.contains_only("user1", "user2", "user3")))

ent.assert_group_by_gid(

2000,

dict(mem=ent.contains_only("user1", "user2", "user3")))

ent.assert_group_by_name(

"group1x",

dict(mem=ent.contains_only("user11", "user12", "user13")))

ent.assert_group_by_gid(

2010,

dict(mem=ent.contains_only("user11", "user12", "user13")))

ent.assert_group_by_name(

"group2x",

dict(mem=ent.contains_only("user21", "user22", "user23")))

ent.assert_group_by_gid(

2020,

dict(mem=ent.contains_only("user21", "user22", "user23")))

def test_getgrnam_membership_with_mc(ldap_conn, sanity_rfc2307):

test_getgrnam_membership(ldap_conn, sanity_rfc2307)

stop_sssd()

test_getgrnam_membership(ldap_conn, sanity_rfc2307)

def assert_user_gids_equal(user, expected_gids):

(res, errno, gids) = sssd_id.get_user_gids(user)

assert res == sssd_id.NssReturnCode.SUCCESS, \

"Could not find groups for user %s, %d" % (user, errno)

assert sorted(gids) == sorted(expected_gids), \

"result: %s\n expected %s" % (

", ".join(["%s" % s for s in sorted(gids)]),

", ".join(["%s" % s for s in sorted(expected_gids)])

)

def test_initgroups(ldap_conn, sanity_rfc2307):

assert_user_gids_equal('user1', [2000, 2001])

assert_user_gids_equal('user2', [2000, 2002])

assert_user_gids_equal('user3', [2000, 2003])

assert_user_gids_equal('user11', [2010, 2001])

assert_user_gids_equal('user12', [2010, 2002])

assert_user_gids_equal('user13', [2010, 2003])

assert_user_gids_equal('user21', [2020, 2001])

assert_user_gids_equal('user22', [2020, 2002])

assert_user_gids_equal('user23', [2020, 2003])

def test_initgroups_with_mc(ldap_conn, sanity_rfc2307):

test_initgroups(ldap_conn, sanity_rfc2307)

stop_sssd()

test_initgroups(ldap_conn, sanity_rfc2307)

def test_initgroups_fqname_with_mc(ldap_conn, fqname_rfc2307):

assert_user_gids_equal('user1@LDAP', [2000, 2001])

stop_sssd()

assert_user_gids_equal('user1@LDAP', [2000, 2001])

def assert_initgroups_equal(user, primary_gid, expected_gids):

(res, errno, gids) = sssd_id.call_sssd_initgroups(user, primary_gid)

assert res == sssd_id.NssReturnCode.SUCCESS, \

"Could not find groups for user %s, %d" % (user, errno)

assert sorted(gids) == sorted(expected_gids), \

"result: %s\n expected %s" % (

", ".join(["%s" % s for s in sorted(gids)]),

", ".join(["%s" % s for s in sorted(expected_gids)])

)

def assert_stored_last_initgroups(user1_case1, user1_case2, user1_case_last,

primary_gid, expected_gids):

assert_initgroups_equal(user1_case1, primary_gid, expected_gids)

assert_initgroups_equal(user1_case2, primary_gid, expected_gids)

assert_initgroups_equal(user1_case_last, primary_gid, expected_gids)

stop_sssd()

user = user1_case1

(res, errno, _) = sssd_id.call_sssd_initgroups(user, primary_gid)

assert res == sssd_id.NssReturnCode.UNAVAIL, \

"Initgroups for user shoudl fail user %s, %d, %d" % (user, res, errno)

user = user1_case2

(res, errno, _) = sssd_id.call_sssd_initgroups(user, primary_gid)

assert res == sssd_id.NssReturnCode.UNAVAIL, \

"Initgroups for user shoudl fail user %s, %d, %d" % (user, res, errno)

# Just last invocation of initgroups shoudl PASS

# Otherwise, we would not be able to invalidate it

assert_initgroups_equal(user1_case_last, primary_gid, expected_gids)

def test_initgroups_case_insensitive_with_mc1(ldap_conn,

fqname_case_insensitive_rfc2307):

user1_case1 = 'User1@LDAP'

user1_case2 = 'uSer1@LDAP'

user1_case_last = 'usEr1@LDAP'

primary_gid = 2001

expected_gids = [2000, 2001]

assert_stored_last_initgroups(user1_case1, user1_case2, user1_case_last,

primary_gid, expected_gids)

def test_initgroups_case_insensitive_with_mc2(ldap_conn,

fqname_case_insensitive_rfc2307):

user1_case1 = 'usEr1@LDAP'

user1_case2 = 'User1@LDAP'

user1_case_last = 'uSer1@LDAP'

primary_gid = 2001

expected_gids = [2000, 2001]

assert_stored_last_initgroups(user1_case1, user1_case2, user1_case_last,

primary_gid, expected_gids)

def test_initgroups_case_insensitive_with_mc3(ldap_conn,

fqname_case_insensitive_rfc2307):

user1_case1 = 'uSer1@LDAP'

user1_case2 = 'usEr1@LDAP'

user1_case_last = 'User1@LDAP'

primary_gid = 2001

expected_gids = [2000, 2001]

assert_stored_last_initgroups(user1_case1, user1_case2, user1_case_last,

primary_gid, expected_gids)

def run_simple_test_with_initgroups():

ent.assert_passwd_by_name(

'user1',

dict(name='user1', passwd='*', uid=1001, gid=2001,

gecos='1001', shell='/bin/bash'))

ent.assert_passwd_by_uid(

1001,

dict(name='user1', passwd='*', uid=1001, gid=2001,

gecos='1001', shell='/bin/bash'))

ent.assert_group_by_name(

"group1",

dict(mem=ent.contains_only("user1", "user11", "user21")))

ent.assert_group_by_gid(

2001,

dict(mem=ent.contains_only("user1", "user11", "user21")))

# unrelated group to user1

ent.assert_group_by_name(

"group2",

dict(mem=ent.contains_only("user2", "user12", "user22")))

ent.assert_group_by_gid(

2002,

dict(mem=ent.contains_only("user2", "user12", "user22")))

assert_initgroups_equal("user1", 2001, [2000, 2001])

def test_invalidation_of_gids_after_initgroups(ldap_conn, sanity_rfc2307):

# the sssd cache was empty and not all user's group were

# resolved with getgr{nm,gid}. Therefore there is a change in

# group membership => user groups should be invalidated

run_simple_test_with_initgroups()

assert_initgroups_equal("user1", 2001, [2000, 2001])

stop_sssd()

ent.assert_passwd_by_name(

'user1',

dict(name='user1', passwd='*', uid=1001, gid=2001,

gecos='1001', shell='/bin/bash'))

ent.assert_passwd_by_uid(

1001,

dict(name='user1', passwd='*', uid=1001, gid=2001,

gecos='1001', shell='/bin/bash'))

# unrelated group to user1 must be returned

ent.assert_group_by_name(

"group2",

dict(mem=ent.contains_only("user2", "user12", "user22")))

ent.assert_group_by_gid(

2002,

dict(mem=ent.contains_only("user2", "user12", "user22")))

assert_initgroups_equal("user1", 2001, [2000, 2001])

# user groups must be invalidated

for group in ["group1", "group0x"]:

with pytest.raises(KeyError):

grp.getgrnam(group)

for gid in [2000, 2001]:

with pytest.raises(KeyError):

grp.getgrgid(gid)

def test_initgroups_without_change_in_membership(ldap_conn, sanity_rfc2307):

# the sssd cache was empty and not all user's group were

# resolved with getgr{nm,gid}. Therefore there is a change in

# group membership => user groups should be invalidated

run_simple_test_with_initgroups()

# invalidate cache

subprocess.call(["sss_cache", "-E"])

# all users and groups will be just refreshed from LDAP

# but there will not be a change in group membership

# user groups should not be invlaidated

run_simple_test_with_initgroups()

stop_sssd()

# everything should be in memory cache

run_simple_test_with_initgroups()

def assert_mc_records_for_user1():

ent.assert_passwd_by_name(

'user1',

dict(name='user1', passwd='*', uid=1001, gid=2001,

gecos='1001', shell='/bin/bash'))

ent.assert_passwd_by_uid(

1001,

dict(name='user1', passwd='*', uid=1001, gid=2001,

gecos='1001', shell='/bin/bash'))

ent.assert_group_by_name(

"group1",

dict(mem=ent.contains_only("user1", "user11", "user21")))

ent.assert_group_by_gid(

2001,

dict(mem=ent.contains_only("user1", "user11", "user21")))

ent.assert_group_by_name(

"group0x",

dict(mem=ent.contains_only("user1", "user2", "user3")))

ent.assert_group_by_gid(

2000,

dict(mem=ent.contains_only("user1", "user2", "user3")))

assert_initgroups_equal("user1", 2001, [2000, 2001])

def assert_missing_mc_records_for_user1():

with pytest.raises(KeyError):

pwd.getpwnam("user1")

with pytest.raises(KeyError):

pwd.getpwuid(1001)

for gid in [2000, 2001]:

with pytest.raises(KeyError):

grp.getgrgid(gid)

for group in ["group0x", "group1"]:

with pytest.raises(KeyError):

grp.getgrnam(group)

(res, err, _) = sssd_id.call_sssd_initgroups("user1", 2001)

assert res == sssd_id.NssReturnCode.UNAVAIL, \

"Initgroups should not find anything after invalidation of mc.\n" \

"User user1, errno:%d" % err

def test_invalidate_user_before_stop(ldap_conn, sanity_rfc2307):

# initialize cache with full ID

(res, errno, _) = sssd_id.get_user_groups("user1")

assert res == sssd_id.NssReturnCode.SUCCESS, \

"Could not find groups for user1 %s, %d" % errno

assert_mc_records_for_user1()

subprocess.call(["sss_cache", "-u", "user1"])

stop_sssd()

assert_missing_mc_records_for_user1()

def test_invalidate_user_after_stop(ldap_conn, sanity_rfc2307):

# initialize cache with full ID

(res, errno, _) = sssd_id.get_user_groups("user1")

assert res == sssd_id.NssReturnCode.SUCCESS, \

"Could not find groups for user1 %s, %d" % errno

assert_mc_records_for_user1()

stop_sssd()

subprocess.call(["sss_cache", "-u", "user1"])

assert_missing_mc_records_for_user1()

def test_invalidate_users_before_stop(ldap_conn, sanity_rfc2307):

# initialize cache with full ID

(res, errno, _) = sssd_id.get_user_groups("user1")

assert res == sssd_id.NssReturnCode.SUCCESS, \

"Could not find groups for user1 %s, %d" % errno

assert_mc_records_for_user1()

subprocess.call(["sss_cache", "-U"])

stop_sssd()

assert_missing_mc_records_for_user1()

def test_invalidate_users_after_stop(ldap_conn, sanity_rfc2307):

# initialize cache with full ID

(res, errno, _) = sssd_id.get_user_groups("user1")

assert res == sssd_id.NssReturnCode.SUCCESS, \

"Could not find groups for user1 %s, %d" % errno

assert_mc_records_for_user1()

stop_sssd()

subprocess.call(["sss_cache", "-U"])

assert_missing_mc_records_for_user1()

def test_invalidate_group_before_stop(ldap_conn, sanity_rfc2307):

# initialize cache with full ID

(res, errno, _) = sssd_id.get_user_groups("user1")

assert res == sssd_id.NssReturnCode.SUCCESS, \

"Could not find groups for user1 %s, %d" % errno

assert_mc_records_for_user1()

subprocess.call(["sss_cache", "-g", "group1"])

stop_sssd()

assert_missing_mc_records_for_user1()

def test_invalidate_group_after_stop(ldap_conn, sanity_rfc2307):

# initialize cache with full ID

(res, errno, _) = sssd_id.get_user_groups("user1")

assert res == sssd_id.NssReturnCode.SUCCESS, \

"Could not find groups for user1 %s, %d" % errno

assert_mc_records_for_user1()

stop_sssd()

subprocess.call(["sss_cache", "-g", "group1"])

assert_missing_mc_records_for_user1()

def test_invalidate_groups_before_stop(ldap_conn, sanity_rfc2307):

# initialize cache with full ID

(res, errno, _) = sssd_id.get_user_groups("user1")

assert res == sssd_id.NssReturnCode.SUCCESS, \

"Could not find groups for user1 %s, %d" % errno

assert_mc_records_for_user1()

subprocess.call(["sss_cache", "-G"])

stop_sssd()

assert_missing_mc_records_for_user1()

def test_invalidate_groups_after_stop(ldap_conn, sanity_rfc2307):

# initialize cache with full ID

(res, errno, _) = sssd_id.get_user_groups("user1")

assert res == sssd_id.NssReturnCode.SUCCESS, \

"Could not find groups for user1 %s, %d" % errno

assert_mc_records_for_user1()

stop_sssd()

subprocess.call(["sss_cache", "-G"])

assert_missing_mc_records_for_user1()

def test_invalidate_everything_before_stop(ldap_conn, sanity_rfc2307):

# initialize cache with full ID

(res, errno, _) = sssd_id.get_user_groups("user1")

assert res == sssd_id.NssReturnCode.SUCCESS, \

"Could not find groups for user1 %s, %d" % errno

assert_mc_records_for_user1()

subprocess.call(["sss_cache", "-E"])

stop_sssd()

assert_missing_mc_records_for_user1()

def test_invalidate_everything_after_stop(ldap_conn, sanity_rfc2307):

# initialize cache with full ID

(res, errno, _) = sssd_id.get_user_groups("user1")

assert res == sssd_id.NssReturnCode.SUCCESS, \

"Could not find groups for user1 %s, %d" % errno

assert_mc_records_for_user1()

stop_sssd()

subprocess.call(["sss_cache", "-E"])

assert_missing_mc_records_for_user1()