How to use cobbler - 10 common examples

"""
        Handle copying of http:// and ftp:// debian repos.
        """

        # warn about not having mirror program.

        mirror_program = "/usr/bin/debmirror"
        if not os.path.exists(mirror_program):
            utils.die(self.logger, "no %s found, please install it" % (mirror_program))

        cmd = ""                  # command to run

        # detect cases that require special handling

        if repo.rpm_list != "" and repo.rpm_list != []:
            utils.die(self.logger, "has_rpm_list not yet supported on apt repos")

        if not repo.arch:
            utils.die(self.logger, "Architecture is required for apt repositories")

        # built destination path for the repo
        dest_path = os.path.join("/var/www/cobbler/repo_mirror", repo.name)

        if repo.mirror_locally:
            # NOTE: Dropping @@suite@@ replace as it is also dropped from
            # from manage_import_debian_ubuntu.py due that repo has no os_version
            # attribute. If it is added again it will break the Web UI!
            # mirror = repo.mirror.replace("@@suite@@",repo.os_version)
            mirror = repo.mirror

            idx = mirror.find("://")
            method = mirror[:idx]

def rsync_sync(self, repo):

        """
        Handle copying of rsync:// and rsync-over-ssh repos.
        """

        if not repo.mirror_locally:
            utils.die(self.logger, "rsync:// urls must be mirrored locally, yum cannot access them directly")

        if repo.rpm_list != "" and repo.rpm_list != []:
            self.logger.warning("--rpm-list is not supported for rsync'd repositories")

        # FIXME: don't hardcode
        dest_path = os.path.join(self.settings.webdir + "/repo_mirror", repo.name)

        spacer = ""
        if not repo.mirror.startswith("rsync://") and not repo.mirror.startswith("/"):
            spacer = "-e ssh"
        if not repo.mirror.strip().endswith("/"):
            repo.mirror = "%s/" % repo.mirror

        # FIXME: wrapper for subprocess that logs to logger
        cmd = "rsync -rltDv --copy-unsafe-links --delete-after %s --delete --exclude-from=/etc/cobbler/rsync.exclude %s %s" % (spacer, pipes.quote(repo.mirror), pipes.quote(dest_path))
        rc = utils.subprocess_call(self.logger, cmd)

def apt_sync(self, repo):

        """
        Handle copying of http:// and ftp:// debian repos.
        """

        # warn about not having mirror program.

        mirror_program = "/usr/bin/debmirror"
        if not os.path.exists(mirror_program):
            utils.die(self.logger, "no %s found, please install it" % (mirror_program))

        cmd = ""                  # command to run

        # detect cases that require special handling

        if repo.rpm_list != "" and repo.rpm_list != []:
            utils.die(self.logger, "has_rpm_list not yet supported on apt repos")

        if not repo.arch:
            utils.die(self.logger, "Architecture is required for apt repositories")

        # built destination path for the repo
        dest_path = os.path.join("/var/www/cobbler/repo_mirror", repo.name)

        if repo.mirror_locally:
            # NOTE: Dropping @@suite@@ replace as it is also dropped from

Write a configuration file for the boot loader(s).
        More system-specific configuration may come in later, if so
        that would appear inside the system object in api.py

        Can be used for different formats, "pxe" (default) and "grub".
        """
        if arch is None:
            raise CX("missing arch")

        if image and not os.path.exists(image.file):
            return None     # nfs:// URLs or something, can't use for TFTP

        if metadata is None:
            metadata = {}

        (rval, settings) = utils.input_string_or_dict(self.settings.to_dict())
        if rval:
            for key in list(settings.keys()):
                metadata[key] = settings[key]
        # ---
        # just some random variables
        template = None
        buffer = ""

        # ---
        autoinstall_path = None
        kernel_path = None
        initrd_path = None
        img_path = None

        if image is None:
            # not image based, it's something normalish

def run_cmd(cmd):
    """
    Run a command

    @param cmd str command
    @return str output
    @raise Exception if return code is not 0
    """

    print("run cmd: %s" % cmd)
    args = shlex.split(cmd)
    (output, rc) = utils.subprocess_sp(None, args, shell=False)
    if rc != 0:
        raise Exception
    return output

self.assertFalse(authorize(self.api, "dne", "sync"), "dne cannot sync")

        # unlike the distro testdistro0, testrepo0 is unowned
        # so any user in the file will be able to edit it.
        for user in [ "admin1", "superlab1", "basement1", "basement2" ]:
           self.assertTrue(authorize(self.api, user, "save_repo", ro), "user %s can save_repo" % user)

        # though dne is still not listed and will be denied
        self.assertFalse(authorize(self.api, "dne", "save_repo", ro), "dne cannot save_repo")

        # if we survive, restore the users file as module testing is done
        if os.path.exists("/tmp/cobbler_ubak"):
           shutil.copyfile("/etc/cobbler/users.conf","/tmp/cobbler_ubak")


class MultiNIC(BootTest):
    
    def test_multi_nic_support(self):


        system = self.api.new_system()
        self.assertTrue(system.set_name("nictest"))
        self.assertTrue(system.set_profile("testprofile0"))
        self.assertTrue(system.set_dns_name("zero","eth0"))
        self.assertTrue(system.set_mac_address("EE:FF:DD:CC:DD:CC","eth1"))
        self.assertTrue(system.set_ip_address("127.0.0.5","eth2"))
        self.assertTrue(system.set_dhcp_tag("zero","eth3"))
        self.assertTrue(system.set_virt_bridge("zero","eth4"))
        self.assertTrue(system.set_gateway("192.168.1.25")) # is global
        self.assertTrue(system.set_name_servers("a.example.org b.example.org")) # is global
        self.assertTrue(system.set_mac_address("AA:AA:BB:BB:CC:CC","eth4"))
        self.assertTrue(system.set_dns_name("fooserver","eth4"))

def test_invalid_delete_distro_would_orphan_profile(self):
        self.failUnlessRaises(CobblerException, self.api.distros().remove, "testdistro0")

    #def test_working_deletes(self):
    #    self.api.clear()
    #    # self.make_basic_config()
    #    #self.assertTrue(self.api.systems().remove("testsystem0"))
    #    self.api.serialize()
    #    self.assertTrue(self.api.remove_profile("testprofile0"))
    #    self.assertTrue(self.api.remove_distro("testdistro0"))
    #    #self.assertFalse(self.api.find_system(name="testsystem0"))
    #    self.assertFalse(self.api.find_profile(name="testprofile0"))
    #    self.assertFalse(self.api.find_distro(name="testdistro0"))

class TestCheck(BootTest):

   def test_check(self):
       # we can't know if it's supposed to fail in advance
       # (ain't that the halting problem), but it shouldn't ever
       # throw exceptions.
       self.api.check()

class TestSync(BootTest):

   def test_real_run(self):
       # syncing a real test run in an automated environment would
       # break a valid cobbler configuration, so we're not going to
       # test this here.
       pass

class TestListings(BootTest):

#    self.api.serialize()
    #    self.assertTrue(self.api.remove_profile("testprofile0"))
    #    self.assertTrue(self.api.remove_distro("testdistro0"))
    #    #self.assertFalse(self.api.find_system(name="testsystem0"))
    #    self.assertFalse(self.api.find_profile(name="testprofile0"))
    #    self.assertFalse(self.api.find_distro(name="testdistro0"))

class TestCheck(BootTest):

   def test_check(self):
       # we can't know if it's supposed to fail in advance
       # (ain't that the halting problem), but it shouldn't ever
       # throw exceptions.
       self.api.check()

class TestSync(BootTest):

   def test_real_run(self):
       # syncing a real test run in an automated environment would
       # break a valid cobbler configuration, so we're not going to
       # test this here.
       pass

class TestListings(BootTest):

   def test_listings(self):
       # check to see if the collection listings output something.
       # this is a minimal check, mainly for coverage, not validity
       # self.make_basic_config()
       self.assertTrue(len(self.api.systems().printable()) > 0)
       self.assertTrue(len(self.api.profiles().printable()) > 0)
       self.assertTrue(len(self.api.distros().printable()) > 0)

pass
        except:
           self.assertTrue(1==2,"wrong exception type")

        # fix the IP and Mac back 
        self.assertTrue(system3.set_ip_address("192.86.75.30","eth6"))
        self.assertTrue(system3.set_mac_address("AE:BE:DE:CE:AE:EE","eth3"))
        # now it works again
        # note that we will not check for duplicate names as we want
        # to test this as an 'edit' operation.
        self.assertTrue(self.api.add_system(system3,check_for_duplicate_names=False,check_for_duplicate_netinfo=True))

        # FIXME: note -- how netinfo is handled when doing renames/copies/edits
        # is more involved and we probably should add tests for that also.

class Ownership(BootTest):

    def test_ownership_params(self):
   
        fd = open("/tmp/test_cobbler_kickstart","w+")
        fd.write("")
        fd.close()

        # find things we are going to test with
        distro = self.api.find_distro(name="testdistro0")
        profile = self.api.find_profile(name="testprofile0")
        system = self.api.find_system(name="testsystem0")
        repo = self.api.find_repo(name="testrepo0")

        # as we didn't specify an owner for objects, the default
        # ownership should be as specified in settings
        default_owner = self.api.settings().default_ownership

def test_system_name_is_an_IP(self):
        system = self.api.new_system()
        name = "192.168.1.54"
        self.assertTrue(system.set_name(name))
        self.assertTrue(system.set_profile("testprofile0"))
        self.assertTrue(self.api.add_system(system))
        self.assertTrue(self.api.find_system(name=name))

    def test_invalid_system_non_referenced_profile(self):
        system = self.api.new_system()
        self.assertTrue(system.set_name("testsystem0"))
        self.failUnlessRaises(CobblerException, system.set_profile, "profiledoesntexist")
        self.failUnlessRaises(CobblerException, self.api.add_system, system)

class SyncContents(BootTest):

    def test_blender_cache_works(self):

        distro = self.api.new_distro()
        self.assertTrue(distro.set_name("D1"))
        self.assertTrue(distro.set_kernel(self.fk_kernel))
        self.assertTrue(distro.set_initrd(self.fk_initrd))
        self.assertTrue(self.api.add_distro(distro))
        self.assertTrue(self.api.find_distro(name="D1"))

        profile = self.api.new_profile()
        self.assertTrue(profile.set_name("P1"))
        self.assertTrue(profile.set_distro("D1"))
        self.assertTrue(profile.set_kickstart("/var/lib/cobbler/kickstarts/sample.ks"))
        self.assertTrue(self.api.add_profile(profile))
        assert self.api.find_profile(name="P1") != None