source4/lib/ldb/tests/python/ldifapplier.py

   1 #!/usr/bin/python
   2
   3 import sys
   4
   5 sys.path.append("bin/python")
   6 sys.path.append("../lib/subunit/python")
   7 sys.path.append("../schema_config_ldifs")
   8
   9 from samba.provision import (read_and_sub_file)
  10 import samba
  11 import ldb
  12 from samba import Ldb, substitute_var, check_all_substituted
  13
  14 class LdifApplier:
  15
  16     def __init__(self, domain, session, credentials, lp):
  17         self.sambaprep_list = ["../schema_config_ldifs/SambaNewAttributes.ldf",
  18                                "../schema_config_ldifs/schemaUpdateNow.ldf",
  19                                "../schema_config_ldifs/SambaNewClasses.ldf",
  20                                "../schema_config_ldifs/schemaUpdateNow.ldf",
  21                                "../schema_config_ldifs/SambaNewConfig.ldf",
  22                                "../schema_config_ldifs/schemaUpdateNow.ldf"
  23                                ]
  24         self.schemamod_list = ["../schema_config_ldifs/newAttributes.ldf",
  25                                "../schema_config_ldifs/changedAttributes.ldf",
  26                                "../schema_config_ldifs/schemaUpdateNow.ldf",
  27                                "../schema_config_ldifs/newClasses.ldf",
  28                                "../schema_config_ldifs/schemaUpdateNow.ldf",
  29                                "../schema_config_ldifs/additionExExtentions.ldf",
  30                                "../schema_config_ldifs/schemaUpdateNow.ldf",
  31                                "../schema_config_ldifs/changedClasses.ldf",
  32                                "../schema_config_ldifs/schemaUpdateNow.ldf"
  33                                ]
  34
  35         self.configmod_list = ["../schema_config_ldifs/baseChanges.ldf",
  36                                "../schema_config_ldifs/schemaUpdateNow.ldf"]
  37
  38         self.test1_list =     [ "../schema_config_ldifs/additionExExtentions1.ldf",
  39                                "../schema_config_ldifs/schemaUpdateNow.ldf"]
  40
  41         self.test2_list =     ["../schema_config_ldifs/SambaNewAttributes.ldf",
  42                                "../schema_config_ldifs/schemaUpdateNow.ldf"]
  43
  44         self.test3_list =     ["../schema_config_ldifs/schemaUpdateNow.ldf"]
  45
  46
  47         self.domaindn = "DC=" + domain.replace(".", ",DC=")
  48
  49         self.conf_context = "CN=Configuration," + self.domaindn
  50         self.schema_context = "CN=Schema," + self.conf_context
  51         self.subst_vars = {"SCHEMADN": self.schema_context,
  52                            "CONFIGDN": self.conf_context }
  53         self.session = session
  54         self.credentials = credentials
  55         self.lp = lp
  56         host = "ldap://localhost:389"
  57
  58         self.ldb = Ldb(host, credentials=self.credentials, session_info=session, lp=self.lp)
  59
  60     def add_or_modify_ldf(self, ldif_path):
  61         assert isinstance(ldif_path, str)
  62
  63         ldif = read_and_sub_file(ldif_path, self.subst_vars)
  64         for changetype, msg in self.ldb.parse_ldif(ldif):
  65             try:
  66                 if changetype == ldb.CHANGETYPE_NONE or changetype == ldb.CHANGETYPE_ADD:
  67                     self.ldb.add(msg)
  68                 else:
  69                     self.ldb.modify(msg)
  70             except ldb.LdbError, (LDB_ERR_ENTRY_ALREADY_EXISTS, _):
  71                 pass
  72
  73     def add_or_modify_ldf_no_ex(self, ldif_path):
  74         assert isinstance(ldif_path, str)
  75
  76         ldif = read_and_sub_file(ldif_path, self.subst_vars)
  77         for changetype, msg in self.ldb.parse_ldif(ldif):
  78             if changetype == ldb.CHANGETYPE_NONE or changetype == ldb.CHANGETYPE_ADD:
  79                 self.ldb.add(msg)
  80             else:
  81                 self.ldb.modify(msg)
  82
  83     def apply_samba(self):
  84         for i in self.sambaprep_list:
  85             print "Applying: %s\n" % i
  86             self.add_or_modify_ldf(i)
  87
  88
  89     def apply_schema(self):
  90         for i in self.schemamod_list:
  91             print "Applying: %s\n" % i
  92             self.add_or_modify_ldf(i)
  93
  94     def apply_config(self):
  95         for i in self.configmod_list:
  96             print "Applying: %s\n" % i
  97             self.add_or_modify_ldf(i)
  98
  99
 100 #todo exception handling, etc.
 101     def apply_all(self):
 102         self.apply_samba()
 103         self.apply_schema()
 104         self.apply_config()
 105         #called twice on purpose
 106         self.apply_config()
 107
 108     def apply_test1(self):
 109         for i in self.test1_list:
 110             print "Applying: %s\n" % i
 111             self.add_or_modify_ldf_no_ex(i)
 112
 113     def apply_test2(self):
 114         for i in self.test2_list:
 115             print "Applying: %s\n" % i
 116             self.add_or_modify_ldf(i)
 117
 118     def apply_test3(self):
 119         for i in self.test3_list:
 120             print "Applying: %s\n" % i
 121             self.add_or_modify_ldf_no_ex(i)