python-cobbler 2.2.2-0ubuntu33 / usr / share / pyshared / cobbler / test_basic.py

# Test cases for Cobbler
#
# Michael DeHaan <mdehaan@redhat.com>

import sys
import unittest
import os
import tempfile
import shutil
import traceback

from cexceptions import *  

import modules.authz_ownership as authz_module
import api
import config
import utils
utils.TEST_MODE = True

FAKE_INITRD="initrd-2.6.15-1.2054_FAKE.img"
FAKE_INITRD2="initrd-2.5.16-2.2055_FAKE.img"
FAKE_INITRD3="initrd-1.8.18-3.9999_FAKE.img"
FAKE_KERNEL="vmlinuz-2.6.15-1.2054_FAKE"
FAKE_KERNEL2="vmlinuz-2.5.16-2.2055_FAKE"
FAKE_KERNEL3="vmlinuz-1.8.18-3.9999_FAKE"

cleanup_dirs = []

class BootTest(unittest.TestCase):

    def setUp(self):


        # Create temp dir
        self.topdir = "/tmp/cobbler_test"
        try:
            os.makedirs(self.topdir)
        except:
            pass

        self.fk_initrd = os.path.join(self.topdir,  FAKE_INITRD)
        self.fk_initrd2 = os.path.join(self.topdir, FAKE_INITRD2)
        self.fk_initrd3 = os.path.join(self.topdir, FAKE_INITRD3)

        self.fk_kernel = os.path.join(self.topdir,  FAKE_KERNEL)
        self.fk_kernel2 = os.path.join(self.topdir, FAKE_KERNEL2)
        self.fk_kernel3 = os.path.join(self.topdir, FAKE_KERNEL3)

        self.api = api.BootAPI()
        create = [ self.fk_initrd, self.fk_initrd2, self.fk_initrd3,
                self.fk_kernel, self.fk_kernel2, self.fk_kernel3 ]
        for fn in create:
            f = open(fn,"w+")
            f.close()
        self.__make_basic_config()

    def tearDown(self):

        for x in self.api.distros():
            self.api.remove_distro(x,recursive=True)
        for y in self.api.repos():
            self.api.remove_repo(y)
        for z in self.api.images():
            self.api.remove_image(z)
        shutil.rmtree(self.topdir,ignore_errors=True)
        self.api = None

    def __make_basic_config(self):
        distro = self.api.new_distro()
        self.assertTrue(distro.set_name("testdistro0"))
        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="testdistro0"))
        
        profile = self.api.new_profile()
        self.assertTrue(profile.set_name("testprofile0"))
        self.assertTrue(profile.set_distro("testdistro0"))
        self.assertTrue(profile.set_kickstart("/var/lib/cobbler/kickstarts/sample_end.ks"))
        self.assertTrue(self.api.add_profile(profile))
        self.assertTrue(self.api.find_profile(name="testprofile0"))

        system = self.api.new_system()
        self.assertTrue(system.set_name("testsystem0"))
        self.assertTrue(system.set_mac_address("BB:EE:EE:EE:EE:FF","eth0"))
        self.assertTrue(system.set_ip_address("192.51.51.50","eth0"))
        self.assertTrue(system.set_profile("testprofile0"))
        self.assertTrue(self.api.add_system(system))
        self.assertTrue(self.api.find_system(name="testsystem0"))

        repo = self.api.new_repo()
        try:
            os.makedirs("/tmp/test_example_cobbler_repo")
        except:
            pass
        fd = open("/tmp/test_example_cobbler_repo/test.file", "w+")
        fd.write("hello!")
        fd.close()
        self.assertTrue(repo.set_name("testrepo0"))
        self.assertTrue(repo.set_mirror("/tmp/test_example_cobbler_repo"))
        self.assertTrue(self.api.add_repo(repo))

        image = self.api.new_image()
        self.assertTrue(image.set_name("testimage0"))
        self.assertTrue(image.set_file(self.fk_initrd)) # meaningless path
        self.assertTrue(self.api.add_image(image))


class RenameTest(BootTest):

    def __tester(self, finder, renamer, name1, name2):

        x = finder(name1)
        assert x is not None

        renamer(x, name2)
        x = finder(name1)
        y = finder(name2)
        assert x is None
        assert y is not None
        
        renamer(y, name1) 
        x = finder(name1)
        y = finder(name2)
        assert x is not None
        assert y is None 

    def test_renames(self):
        self.__tester(self.api.find_distro, self.api.rename_distro, "testdistro0", "testdistro1")
        self.__tester(self.api.find_profile, self.api.rename_profile, "testprofile0", "testprofile1")
        self.__tester(self.api.find_system, self.api.rename_system, "testsystem0", "testsystem1")
        self.__tester(self.api.find_repo, self.api.rename_repo, "testrepo0", "testrepo1")
        self.__tester(self.api.find_image, self.api.rename_image, "testimage0", "testimage1")


class DuplicateNamesAndIpPrevention(BootTest):

    """
    The command line (and WebUI) have checks to prevent new system
    additions from conflicting with existing systems and overwriting
    them inadvertantly. This class tests that code.  NOTE: General API
    users will /not/ encounter these checks.
    """

    def test_duplicate_prevention(self):

        # find things we are going to test with
        distro1 = self.api.find_distro(name="testdistro0")
        profile1 = self.api.find_profile(name="testprofile0")
        system1 = self.api.find_system(name="testsystem0")
        repo1 = self.api.find_repo(name="testrepo0")

        # make sure we can't overwrite a previous distro with
        # the equivalent of an "add" (not an edit) on the
        # command line.
        distro2 = self.api.new_distro()
        self.assertTrue(distro2.set_name("testdistro0"))
        self.assertTrue(distro2.set_kernel(self.fk_kernel))
        self.assertTrue(distro2.set_initrd(self.fk_initrd))
        self.assertTrue(distro2.set_owners("canary"))
        # this should fail
        try:
           self.api.add_distro(distro2,check_for_duplicate_names=True)
           self.assertTrue(1==2,"distro add should fail")
        except CobblerException:
           pass
        except:
           self.assertTrue(1==2,"exception type")
        # we caught the exception but make doubly sure there was no write
        distro_check = self.api.find_distro(name="testdistro0")
        self.assertTrue("canary" not in distro_check.owners)

        # repeat the check for profiles
        profile2 = self.api.new_profile()
        self.assertTrue(profile2.set_name("testprofile0"))
        self.assertTrue(profile2.set_distro("testdistro0"))
        # this should fail
        try:
            self.api.add_profile(profile2,check_for_duplicate_names=True)
            self.assertTrue(1==2,"profile add should fail")
        except CobblerException:
            pass
        except:
            traceback.print_exc()
            self.assertTrue(1==2,"exception type")

        # repeat the check for systems (just names this time)
        system2 = self.api.new_system()
        self.assertTrue(system2.set_name("testsystem0"))
        self.assertTrue(system2.set_profile("testprofile0"))
        # this should fail
        try:
            self.api.add_system(system2,check_for_duplicate_names=True)
            self.assertTrue(1==2,"system add should fail")
        except CobblerException:
            pass
        except:
            traceback.print_exc()
            self.assertTrue(1==2,"exception type")

        # repeat the check for repos
        repo2 = self.api.new_repo()
        self.assertTrue(repo2.set_name("testrepo0"))
        self.assertTrue(repo2.set_mirror("http://imaginary"))
        # self.failUnlessRaises(CobblerException,self.api.add_repo,[repo,check_for_duplicate_names=True])
        try:
            self.api.add_repo(repo2,check_for_duplicate_names=True)
            self.assertTrue(1==2,"repo add should fail")
        except CobblerException:
            pass
        except:
            self.assertTrue(1==2,"exception type")

        # now one more check to verify we can't add a system
        # of a different name but duplicate netinfo.  
        system3 = self.api.new_system()
        self.assertTrue(system3.set_name("unused_name"))
        self.assertTrue(system3.set_profile("testprofile0"))
        # MAC is initially accepted
        self.assertTrue(system3.set_mac_address("BB:EE:EE:EE:EE:FF","eth3"))
        # can't add as this MAC already exists!  

        #self.failUnlessRaises(CobblerException,self.api.add_system,[system3,check_for_duplicate_names=True,check_for_duplicate_netinfo=True)
        try:
           self.api.add_system(system3,check_for_duplicate_names=True,check_for_duplicate_netinfo=True)
        except CobblerException: 
           pass
        except:
           traceback.print_exc()
           self.assertTrue(1==2,"wrong exception type")

        # set the MAC to a different value and try again
        self.assertTrue(system3.set_mac_address("FF:EE:EE:EE:EE:DD","eth3"))
        # it should work
        self.assertTrue(self.api.add_system(system3,check_for_duplicate_names=False,check_for_duplicate_netinfo=True))
        # now set the IP so that collides
        self.assertTrue(system3.set_ip_address("192.51.51.50","eth6"))
        # this should also fail

        # self.failUnlessRaises(CobblerException,self.api.add_system,[system3,check_for_duplicate_names=True,check_for_duplicate_netinfo=True)
        try:
           self.api.add_system(system3,check_for_duplicate_names=True,check_for_duplicate_netinfo=True)
           self.assertTrue(1==2,"system add should fail")
        except CobblerException:
           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
        for obj in [ distro, profile, system, repo ]:
            self.assertTrue(obj is not None)
            self.assertEquals(obj.owners, default_owner, "default owner for %s" % obj)        

        # verify we can test things
        self.assertTrue(distro.set_owners(["superlab","basement1"]))
        self.assertTrue(profile.set_owners(["superlab","basement1"]))
        self.assertTrue(profile.set_kickstart("/tmp/test_cobbler_kickstart"))
        self.assertTrue(system.set_owners(["superlab","basement1","basement3"]))
        self.assertTrue(repo.set_owners([]))
        self.api.add_distro(distro)
        self.api.add_profile(profile)
        self.api.add_system(system)
        self.api.add_repo(repo)

        # now edit the groups file.  We won't test the full XMLRPC
        # auth stack here, but just the module in question

        def authorize(api, user, resource, arg1=None, arg2=None):
            return authz_module.authorize(api, user,resource,arg1,arg2)

        # if the users.conf file exists, back it up for the tests
        if os.path.exists("/etc/cobbler/users.conf"):
           shutil.copyfile("/etc/cobbler/users.conf","/tmp/cobbler_ubak")
         
        fd = open("/etc/cobbler/users.conf","w+")
        fd.write("\n")
        fd.write("[admins]\n")
        fd.write("admin1 = 1\n")
        fd.write("\n")
        fd.write("[superlab]\n")
        fd.write("superlab1 = 1\n")
        fd.write("superlab2 = 1\n")
        fd.write("\n")
        fd.write("[basement]\n") 
        fd.write("basement1 = 1\n")      
        fd.write("basement2 = 1\n")      
        fd.write("basement3 = 1\n")      
        fd.close()


        xo = self.api.find_distro("testdistro0")
        xn = "testdistro0"
        ro = self.api.find_repo("testrepo0")
        rn = "testrepo0"

        # WARNING: complex test explanation follows! 
        # we must ensure those who can edit the kickstart are only those
        # who can edit all objects that depend on the said kickstart
        # in this test, superlab & basement1 can edit test_profile0
        # superlab & basement1/3 can edit test_system0
        # the systems share a common kickstart record (in this case
        # explicitly set, which is a bit arbitrary as they are parent/child
        # nodes, but the concept is not limited to this).  
        # Therefore the correct result is that the following users can edit:
        #    admin1, superlab1, superlab2
        # And these folks can't
        #    basement1, basement2
        # Basement2 is rejected because the kickstart is shared by something
        # basmeent2 can not edit.


        for user in [ "admin1", "superlab1", "superlab2", "basement1" ]:
           self.assertTrue(authorize(self.api, user, "write_kickstart", "/tmp/test_cobbler_kickstart"), "%s can modify_kickstart" % user)

        for user in [ "basement2", "dne" ]:
           self.assertFalse(authorize(self.api, user, "write_kickstart", "/tmp/test_cobbler_kickstart"), "%s can modify_kickstart" % user)

        # ensure admin1 can edit (he's an admin) and do other tasks
        # same applies to basement1 who is explicitly added as a user
        # and superlab1 who is in a group in the ownership list
        for user in ["admin1","superlab1","basement1"]:
           self.assertTrue(authorize(self.api, user, "save_distro", xo),"%s can save_distro" % user)
           self.assertTrue(authorize(self.api, user, "modify_distro", xo),"%s can modify_distro" % user)
           self.assertTrue(authorize(self.api, user, "copy_distro", xo),"%s can copy_distro" % user)
           self.assertTrue(authorize(self.api, user, "remove_distro", xn),"%s can remove_distro" % user)  

        # ensure all users in the file can sync
        for user in [ "admin1", "superlab1", "basement1", "basement2" ]:     
           self.assertTrue(authorize(self.api, user, "sync"))

        # make sure basement2 can't edit (not in group)
        # and same goes for "dne" (does not exist in users.conf)
        
        for user in [ "basement2", "dne" ]:
           self.assertFalse(authorize(self.api, user, "save_distro", xo), "user %s cannot save_distro" % user)
           self.assertFalse(authorize(self.api, user, "modify_distro", xo), "user %s cannot modify_distro" % user)
           self.assertFalse(authorize(self.api, user, "remove_distro", xn), "user %s cannot remove_distro" % user)
 
        # basement2 is in the file so he can still copy
        self.assertTrue(authorize(self.api, "basement2", "copy_distro", xo), "basement2 can copy_distro")

        # dne can not copy or sync either (not in the users.conf)
        self.assertFalse(authorize(self.api, "dne", "copy_distro", xo), "dne cannot copy_distro")
        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"))
        self.assertTrue(system.set_dhcp_tag("red","eth4"))
        self.assertTrue(system.set_ip_address("192.168.1.26","eth4"))
        self.assertTrue(system.set_netmask("255.255.255.0","eth4"))
        self.assertTrue(system.set_dhcp_tag("tag2","eth5"))
        self.assertTrue(self.api.add_system(system))
        self.assertTrue(self.api.find_system(dns_name="fooserver"))
        self.assertTrue(self.api.find_system(mac_address="EE:FF:DD:CC:DD:CC"))
        self.assertTrue(self.api.find_system(ip_address="127.0.0.5"))
        self.assertTrue(self.api.find_system(virt_bridge="zero"))
        self.assertTrue(self.api.find_system(gateway="192.168.1.25"))
        self.assertTrue(self.api.find_system(netmask="255.255.255.0"))
        self.assertTrue(self.api.find_system(dhcp_tag="tag2"))
        self.assertTrue(self.api.find_system(dhcp_tag="zero"))

        # verify that systems has exactly 5 interfaces
        self.assertTrue(len(system.interfaces.keys()) == 6)

        # now check one interface to make sure it's exactly right
        # and we didn't accidentally fill in any other fields elsewhere

        self.assertTrue(system.interfaces.has_key("eth4"))
        self.assertTrue(system.gateway == "192.168.1.25")
        for (name,intf) in system.interfaces.iteritems():
            if name == "eth4": # xmlrpc dicts must have string keys, so we must also
                self.assertTrue(intf["virt_bridge"] == "zero")
                self.assertTrue(intf["netmask"] == "255.255.255.0")
                self.assertTrue(intf["mac_address"] == "AA:AA:BB:BB:CC:CC")
                self.assertTrue(intf["ip_address"] == "192.168.1.26")
                self.assertTrue(intf["dns_name"] == "fooserver")
                self.assertTrue(intf["dhcp_tag"] == "red")
 
class Utilities(BootTest):

    def _expeq(self, expected, actual):
        try:
            self.failUnlessEqual(expected, actual,
                "Expected: %s; actual: %s" % (expected, actual))
        except:
            self.fail("exception during failUnlessEqual")

    def test_kernel_scan(self):
        self.assertTrue(utils.find_kernel(self.fk_kernel))
        self.assertFalse(utils.find_kernel("filedoesnotexist"))
        self._expeq(self.fk_kernel, utils.find_kernel(self.topdir))

    def test_initrd_scan(self):
        self.assertTrue(utils.find_initrd(self.fk_initrd))
        self.assertFalse(utils.find_initrd("filedoesnotexist"))
        self._expeq(self.fk_initrd, utils.find_initrd(self.topdir))

    def test_kickstart_scan(self):
        # we don't check to see if kickstart files look like anything
        # so this will pass
        self.assertTrue(utils.find_kickstart("filedoesnotexist") is None)
        self.assertTrue(utils.find_kickstart(self.topdir) == None)
        self.assertTrue(utils.find_kickstart("http://bar"))
        self.assertTrue(utils.find_kickstart("ftp://bar"))
        self.assertTrue(utils.find_kickstart("nfs://bar"))
        self.assertFalse(utils.find_kickstart("gopher://bar"))

    def test_matching(self):
        self.assertTrue(utils.is_mac("00:C0:B7:7E:55:50"))
        self.assertTrue(utils.is_mac("00:c0:b7:7E:55:50"))
        self.assertFalse(utils.is_mac("00.D0.B7.7E.55.50"))
        self.assertFalse(utils.is_mac("testsystem0"))
        self.assertTrue(utils.is_ip("127.0.0.1"))
        self.assertTrue(utils.is_ip("192.168.1.1"))
        self.assertFalse(utils.is_ip("00:C0:B7:7E:55:50"))
        self.assertFalse(utils.is_ip("testsystem0"))

    def test_some_random_find_commands(self):
        # initial setup...
        self.test_system_name_is_a_MAC()
        # search for a parameter that isn't real, want an error
        self.failUnlessRaises(CobblerException,self.api.systems().find, pond="mcelligots")

        # verify that even though we have several different NICs search still works
        # FIMXE: temprorarily disabled
        # self.assertTrue(self.api.find_system(name="nictest") is not None)

        # search for a parameter with a bad value, want None
        self.assertFalse(self.api.systems().find(name="horton"))
        # one valid parameter another invalid is still an error
        self.failUnlessRaises(CobblerException,self.api.systems().find, name="onefish",pond="mcelligots")
        # searching with no args is ALSO an error
        self.failUnlessRaises(CobblerException, self.api.systems().find)
        # searching for a list returns a list of correct length
        self.assertTrue(len(self.api.systems().find(mac_address="00:16:41:14:B7:71",return_list=True))==1)
        # make sure we can still search without an explicit keyword arg
        self.assertTrue(len(self.api.systems().find("00:16:41:14:B7:71",return_list=True))==1)
        self.assertTrue(self.api.systems().find("00:16:41:14:B7:71"))

    def test_invalid_distro_non_referenced_kernel(self):
        distro = self.api.new_distro()
        self.assertTrue(distro.set_name("testdistro2"))
        self.failUnlessRaises(CobblerException,distro.set_kernel,"filedoesntexist")
        self.assertTrue(distro.set_initrd(self.fk_initrd))
        self.failUnlessRaises(CobblerException, self.api.add_distro, distro)
        self.assertFalse(self.api.distros().find(name="testdistro2"))

    def test_invalid_distro_non_referenced_initrd(self):
        distro = self.api.new_distro()
        self.assertTrue(distro.set_name("testdistro3"))
        self.assertTrue(distro.set_kernel(self.fk_kernel))
        self.failUnlessRaises(CobblerException, distro.set_initrd, "filedoesntexist")
        self.failUnlessRaises(CobblerException, self.api.add_distro, distro)
        self.assertFalse(self.api.distros().find(name="testdistro3"))

    def test_invalid_profile_non_referenced_distro(self):
        profile = self.api.new_profile()
        self.assertTrue(profile.set_name("testprofile11"))
        self.failUnlessRaises(CobblerException, profile.set_distro, "distrodoesntexist")
        self.assertTrue(profile.set_kickstart("/var/lib/cobbler/kickstarts/sample.ks"))
        self.failUnlessRaises(CobblerException, self.api.add_profile, profile)
        self.assertFalse(self.api.profiles().find(name="testprofile2"))

    def test_invalid_profile_kickstart_not_url(self):
        profile = self.api.new_profile()
        self.assertTrue(profile.set_name("testprofile12"))
        self.assertTrue(profile.set_distro("testdistro0"))
        self.failUnlessRaises(CobblerException, profile.set_kickstart, "kickstartdoesntexist")
        # since kickstarts are optional, you can still add it
        self.assertTrue(self.api.add_profile(profile))
        self.assertTrue(self.api.profiles().find(name="testprofile12"))
        # now verify the other kickstart forms would still work
        self.assertTrue(profile.set_kickstart("http://bar"))
        self.assertTrue(profile.set_kickstart("ftp://bar"))
        self.assertTrue(profile.set_kickstart("nfs://bar"))

    def test_profile_virt_parameter_checking(self):
        profile = self.api.new_profile()
        self.assertTrue(profile.set_name("testprofile12b"))
        self.assertTrue(profile.set_distro("testdistro0"))
        self.assertTrue(profile.set_kickstart("http://127.0.0.1/foo"))
        self.assertTrue(profile.set_virt_bridge("xenbr1"))
        self.assertTrue(profile.set_disk_driver("qcow"))
        # sizes must be integers
        self.assertTrue(profile.set_virt_file_size("54321"))
        self.failUnlessRaises(Exception, profile.set_virt_file_size, "huge")
        self.failUnlessRaises(Exception, profile.set_virt_file_size, "54.321")
        # cpus must be integers
        self.assertTrue(profile.set_virt_cpus("2"))
        self.failUnlessRaises(Exception, profile.set_virt_cpus, "3.14")
        self.failUnlessRaises(Exception, profile.set_virt_cpus, "6.02*10^23")
        self.assertTrue(self.api.add_profile(profile))

    def test_inheritance_and_variable_propogation(self):

        # STEP ONE: verify that non-inherited objects behave
        # correctly with ks_meta (we picked this attribute
        # because it's a hash and it's a bit harder to handle
        # than strings).  It should be passed down the render
        # tree to all subnodes

        repo = self.api.new_repo()
        try:
            os.makedirs("/tmp/test_cobbler_repo")
        except:
            pass
        fd = open("/tmp/test_cobbler_repo/test.file", "w+")
        fd.write("hello!")
        fd.close()
        self.assertTrue(repo.set_name("testrepo"))
        self.assertTrue(repo.set_mirror("/tmp/test_cobbler_repo"))
        self.assertTrue(self.api.add_repo(repo))

        profile = self.api.new_profile()
        self.assertTrue(profile.set_name("testprofile12b2"))
        self.assertTrue(profile.set_distro("testdistro0"))
        self.assertTrue(profile.set_kickstart("http://127.0.0.1/foo"))
        self.assertTrue(profile.set_repos(["testrepo"]))
        self.assertTrue(profile.set_name_servers(["asdf"]))
        self.assertTrue(self.api.add_profile(profile))

        # disable this test as it's not a valid repo yet
        # self.api.reposync()

        self.api.sync()
        system = self.api.new_system()
        self.assertTrue(system.set_name("foo"))
        self.assertTrue(system.set_profile("testprofile12b2"))
        self.assertTrue(system.set_ksmeta({"asdf" : "jkl" }))
        self.assertTrue(self.api.add_system(system))
        profile = self.api.profiles().find("testprofile12b2")
        ksmeta = profile.ks_meta
        self.assertFalse(ksmeta.has_key("asdf"))

        # FIXME: do the same for inherited profiles
        # now verify the same for an inherited profile
        # and this time walk up the tree to verify it wasn't
        # applied to any other object except the base.

        profile2 = self.api.new_profile(is_subobject=True)
        profile2.set_name("testprofile12b3")
        profile2.set_parent("testprofile12b2")
        self.api.add_profile(profile2)
        # disable this test as syncing an invalid repo will fail
        # self.api.reposync()
        self.api.sync()

        # FIXME: now add a system to the inherited profile
        # and set a attribute on it that we will later check for

        system2 = self.api.new_system()
        self.assertTrue(system2.set_name("foo2"))
        self.assertTrue(system2.set_profile("testprofile12b3"))
        self.assertTrue(system2.set_ksmeta({"narf" : "troz"}))
        self.assertTrue(self.api.add_system(system2))
        # disable this test as invalid repos don't sync
        # self.api.reposync()
        self.api.sync()

        # FIXME: now evaluate the system object and make sure  
        # that it has inherited the repos value from the superprofile
        # above it's actual profile.  This should NOT be present in the
        # actual object, which we have not modified yet.

        data = utils.blender(self.api, False, system2)
        self.assertTrue(data["repos"] == ["testrepo"])
        self.assertTrue(self.api.profiles().find(system2.profile).repos == "<<inherit>>")

        # now if we set the repos object of the system to an additional
        # repo we should verify it now contains two repos.
        # (FIXME)
        
        repo2 = self.api.new_repo()
        try:
           os.makedirs("/tmp/cobbler_test/repo0")
        except:
           pass
        fd = open("/tmp/cobbler_test/repo0/file.test","w+")
        fd.write("Hi!")
        fd.close()
        self.assertTrue(repo2.set_name("testrepo2"))
        self.assertTrue(repo2.set_mirror("/tmp/cobbler_test/repo0"))
        self.assertTrue(self.api.add_repo(repo2))
        profile2 = self.api.profiles().find("testprofile12b3")
        # note: side check to make sure we can also set to string values
        profile2.set_repos("testrepo2")       
        self.api.add_profile(profile2) # save it 

        # random bug testing: run sync several times and ensure cardinality doesn't change
        #self.api.reposync()
        self.api.sync()
        self.api.sync()
        self.api.sync()

        data = utils.blender(self.api, False, system2)
        self.assertTrue("testrepo" in data["repos"])
        self.assertTrue("testrepo2" in data["repos"])
        self.assertTrue(len(data["repos"]) == 2)
        self.assertTrue(self.api.profiles().find(system2.profile).repos == ["testrepo2"])

        # now double check that the parent profile still only has one repo in it.
        # this is part of our test against upward propogation

        profile = self.api.profiles().find("testprofile12b2")
        self.assertTrue(len(profile.repos) == 1)
        self.assertTrue(profile.repos == ["testrepo"])

        # now see if the subprofile does NOT have the ksmeta attribute
        # this is part of our test against upward propogation

        profile2 = self.api.profiles().find("testprofile12b3")
        self.assertTrue(type(profile2.ks_meta) == type(""))
        self.assertTrue(profile2.ks_meta == "<<inherit>>")

        # now see if the profile above this profile still doesn't have it

        profile = self.api.profiles().find("testprofile12b2")
        self.assertTrue(type(profile.ks_meta) == type({}))
        # self.api.reposync()
        self.api.sync()
        self.assertFalse(profile.ks_meta.has_key("narf"), "profile does not have the system ksmeta")

        #self.api.reposync()
        self.api.sync()

        # verify that the distro did not acquire the property
        # we just set on the leaf system
        distro = self.api.distros().find("testdistro0")
        self.assertTrue(type(distro.ks_meta) == type({}))
        self.assertFalse(distro.ks_meta.has_key("narf"), "distro does not have the system ksmeta")

        # STEP THREE: verify that inheritance appears to work    
        # by setting ks_meta on the subprofile and seeing
        # if it appears on the leaf system ... must use
        # blender functions

        profile2 = self.api.profiles().find("testprofile12b3")
        profile2.set_ksmeta({"canyouseethis" : "yes" })
        self.assertTrue(self.api.add_profile(profile2))
        system2 = self.api.systems().find("foo2")
        data = utils.blender(self.api, False, system2)
        self.assertTrue(data.has_key("ks_meta"))
        self.assertTrue(data["ks_meta"].has_key("canyouseethis"))
        
        # STEP FOUR: do the same on the superprofile and see
        # if that propogates
        
        profile = self.api.profiles().find("testprofile12b2")
        profile.set_ksmeta({"canyouseethisalso" : "yes" })
        self.assertTrue(self.api.add_profile(profile))
        system2 = self.api.systems().find("foo2")
        data = utils.blender(self.api, False, system2)
        self.assertTrue(data.has_key("ks_meta"))
        self.assertTrue(data["ks_meta"].has_key("canyouseethisalso"))

        # STEP FIVE: see if distro attributes propogate

        distro = self.api.distros().find("testdistro0")
        distro.set_ksmeta({"alsoalsowik" : "moose" })
        self.assertTrue(self.api.add_distro(distro))
        system2 = self.api.find_system("foo2")
        data = utils.blender(self.api, False, system2)
        self.assertTrue(data.has_key("ks_meta"))
        self.assertTrue(data["ks_meta"].has_key("alsoalsowik"))
        
        


        # STEP SEVEN:  see if settings changes also propogate
        # TBA 

    def test_system_name_is_a_MAC(self):
        system = self.api.new_system()
        name = "00:16:41:14:B7:71"
        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))
        self.assertTrue(self.api.find_system(mac_address="00:16:41:14:B7:71"))
        self.assertFalse(self.api.find_system(mac_address="thisisnotamac"))

    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

        system = self.api.new_system()
        self.assertTrue(system.set_name("S1"))
        self.assertTrue(system.set_mac_address("BB:EE:EE:EE:EE:FF","eth0"))
        self.assertTrue(system.set_profile("P1"))
        self.assertTrue(self.api.add_system(system))
        assert self.api.find_system(name="S1") != None

        # ensure that the system after being added has the right template data
        # in /tftpboot

        converted="01-bb-ee-ee-ee-ee-ff"

        if os.path.exists("/var/lib/tftpboot"):
            fh = open("/var/lib/tftpboot/pxelinux.cfg/%s" % converted)
        else:
            fh = open("/tftpboot/pxelinux.cfg/%s" % converted)

        data = fh.read()
        self.assertTrue(data.find("/op/ks/") != -1)
        fh.close()

        # ensure that after sync is applied, the blender cache still allows
        # the system data to persist over the profile data in /tftpboot
        # (which was an error we had in 0.6.3)

        self.api.sync()
        if os.path.exists("/var/lib/tftpboot"):
            fh = open("/var/lib/tftpboot/pxelinux.cfg/%s" % converted)
        else:
            fh = open("/tftpboot/pxelinux.cfg/%s" % converted)
        data = fh.read()
        print "DEBUG DATA: %s" % data
        self.assertTrue(data.find("/op/ks/") != -1)
        fh.close()


class Deletions(BootTest):

    #def test_invalid_delete_profile_doesnt_exist(self):
    #    self.failUnlessRaises(CobblerException, self.api.profiles().remove, "doesnotexist")

    def test_invalid_delete_profile_would_orphan_systems(self):
        self.failUnlessRaises(CobblerException, self.api.profiles().remove, "testprofile0")

    #def test_invalid_delete_system_doesnt_exist(self):
    #    self.failUnlessRaises(CobblerException, self.api.systems().remove, "doesnotexist")

    #def test_invalid_delete_distro_doesnt_exist(self):
    #    self.failUnlessRaises(CobblerException, self.api.distros().remove, "doesnotexist")

    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):

   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)

class TestImage(BootTest):

    def test_image_file(self):
        # ensure that only valid names are accepted and invalid ones are rejected
        image = self.api.new_image()
        self.assertTrue(image.set_file("nfs://hostname/path/to/filename.iso"))
        self.assertTrue(image.set_file("nfs://mcpierce@hostname:/path/to/filename.iso"))
        self.assertTrue(image.set_file("nfs://hostname:/path/to/filename.iso"))
        self.assertTrue(image.set_file("nfs://hostname/filename.iso"))
        self.assertTrue(image.set_file("hostname:/path/to/the/filename.iso"))
        self.failUnlessRaises(CX, image.set_file, "hostname:filename.iso")
        self.failUnlessRaises(CX, image.set_file, "path/to/filename.iso")
        self.failUnlessRaises(CX, image.set_file, "hostname:")
        # port is not allowed
        self.failUnlessRaises(CX, image.set_file, "nfs://hostname:1234/path/to/the/filename.iso")

#class TestCLIBasic(BootTest):
#
#   def test_cli(self):
#       # just invoke the CLI to increase coverage and ensure
#       # nothing major is broke at top level.  Full CLI command testing
#       # is not included (yet) since the API tests hit that fairly throughly
#       # and it would easily double the length of the tests.
#       app = "/usr/bin/python"
#       self.assertTrue(subprocess.call([app,"cobbler/cobbler.py","list"]) == 0)

if __name__ == "__main__":
    if not os.path.exists("setup.py"):
        print "tests: must invoke from top level directory"
        sys.exit(1)
    loader = unittest.defaultTestLoader
    test_module = __import__("tests")  # self import considered harmful?
    tests = loader.loadTestsFromModule(test_module)
    runner = unittest.TextTestRunner()
    runner.run(tests)
    sys.exit(0)