5
6
from smart.control import Control
6
7
from smart.cache import Provides
8
from smart.const import NEVER
10
from twisted.internet import reactor
11
from twisted.internet.defer import Deferred
12
from twisted.internet.utils import getProcessOutputAndValue
10
16
from landscape.package.facade import (
11
SmartFacade, TransactionError, DependencyError, SmartError)
17
TransactionError, DependencyError, ChannelError, SmartError,
18
make_apt_deb_channel, make_deb_dir_channel)
13
20
from landscape.tests.mocker import ANY
14
21
from landscape.tests.helpers import LandscapeTest
79
86
pkg = self.facade.get_packages_by_name("name2")[0]
80
87
self.assertEquals(self.facade.get_package_hash(pkg), HASH2)
89
def test_get_package_hashes(self):
90
self.facade.reload_channels()
91
hashes = self.facade.get_package_hashes()
92
self.assertEquals(sorted(hashes), sorted([HASH1, HASH2, HASH3]))
82
94
def test_get_package_by_hash(self):
83
95
self.facade.reload_channels()
84
96
pkg = self.facade.get_package_by_hash(HASH1)
394
405
self.facade.reload_channels()
395
406
self.assertEquals(self.facade.get_package_hash(pkg), None)
408
def test_reset_add_get_channels(self):
410
channels = [("alias0", {"type": "test"}),
411
("alias1", {"type": "test"})]
413
self.facade.reset_channels()
415
self.assertEquals(self.facade.get_channels(), {})
417
self.facade.add_channel(*channels[0])
418
self.facade.add_channel(*channels[1])
420
self.assertEquals(self.facade.get_channels(), dict(channels))
422
def test_make_channels(self):
424
channel0 = make_apt_deb_channel("http://my.url/dir", "hardy", "main")
425
channel1 = make_deb_dir_channel("/my/repo")
427
self.assertEquals(channel0, {"baseurl": "http://my.url/dir",
428
"distribution": "hardy",
429
"components": "main",
432
self.assertEquals(channel1, {"path": "/my/repo",
435
def test_get_arch(self):
437
The L{SmartFacade.get_arch} should return the system dpkg
440
deferred = Deferred()
443
result = getProcessOutputAndValue("/usr/bin/dpkg",
444
("--print-architecture",))
445
def callback((out, err, code)):
446
self.assertEquals(self.facade.get_arch(), out.strip())
447
result.addCallback(callback)
448
result.chainDeferred(deferred)
450
reactor.callWhenRunning(do_test)
453
def test_set_arch_multiple_times(self):
455
repository_dir = os.path.join(os.path.dirname(__file__), "repository")
458
channel = {"baseurl": "file://%s" % repository_dir,
459
"distribution": "hardy",
460
"components": "main",
463
self.facade.set_arch("i386")
464
self.facade.reset_channels()
465
self.facade.add_channel(alias, channel)
466
self.facade.reload_channels()
468
pkgs = self.facade.get_packages()
469
self.assertEquals(len(pkgs), 1)
470
self.assertEquals(pkgs[0].name, "syslinux")
473
self.facade.set_arch("amd64")
474
self.facade.reset_channels()
475
self.facade.add_channel(alias, channel)
476
self.facade.reload_channels()
478
pkgs = self.facade.get_packages()
479
self.assertEquals(len(pkgs), 1)
480
self.assertEquals(pkgs[0].name, "libclthreads2")
482
def test_set_caching_with_reload_error(self):
485
channel = {"type": "deb-dir",
486
"path": "/does/not/exist"}
488
self.facade.reset_channels()
489
self.facade.add_channel(alias, channel)
490
self.facade.set_caching(NEVER)
492
self.assertRaises(ChannelError, self.facade.reload_channels)
493
self.facade._channels = {}
495
ignore_re = re.compile("\[Smart\].*'alias'.*/does/not/exist")
497
self.log_helper.ignored_exception_regexes = [ignore_re]
499
def test_init_landscape_plugins(self):
501
The landscape plugin which helps managing proxies is loaded when smart
502
is initialized: this sets a smart configuration variable and load the
505
self.facade.reload_channels()
506
self.assertTrue(smart.sysconf.get("use-landscape-proxies"))
507
self.assertIn("smart.plugins.landscape", sys.modules)