src/python-lxc/examples/api_test.py

   1 #!/usr/bin/env python3
   2 # -*- coding: utf-8 -*-
   3 #
   4 # api_test.py: Test/demo of the python3-lxc API
   5 #
   6 # (C) Copyright Canonical Ltd. 2012
   7 #
   8 # Authors:
   9 # Stéphane Graber <stgraber@ubuntu.com>
  10 #
  11 # This library is free software; you can redistribute it and/or
  12 # modify it under the terms of the GNU Lesser General Public
  13 # License as published by the Free Software Foundation; either
  14 # version 2.1 of the License, or (at your option) any later version.
  15 #
  16 # This library is distributed in the hope that it will be useful,
  17 # but WITHOUT ANY WARRANTY; without even the implied warranty of
  18 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU
  19 # Lesser General Public License for more details.
  20 #
  21 # You should have received a copy of the GNU Lesser General Public
  22 # License along with this library; if not, write to the Free Software
  23 # Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301
  24 # USA
  25 #
  26
  27 import lxc
  28 import uuid
  29 import os
  30 import subprocess
  31 import sys
  32 import time
  33
  34 # Let's pick a random name, avoiding clashes
  35 CONTAINER_NAME = str(uuid.uuid1())
  36 CLONE_NAME = str(uuid.uuid1())
  37 RENAME_NAME = str(uuid.uuid1())
  38
  39 ## Instantiate the container instance
  40 print("Getting instance for '%s'" % CONTAINER_NAME)
  41 container = lxc.Container(CONTAINER_NAME)
  42
  43 # A few basic checks of the current state
  44 assert(container.config_file_name == "%s/%s/config" %
  45        (lxc.default_config_path, CONTAINER_NAME))
  46 assert(not container.defined)
  47 assert(container.init_pid == -1)
  48 assert(container.name == CONTAINER_NAME)
  49 assert(not container.running)
  50 assert(container.state == "STOPPED")
  51
  52 # Try to get the host architecture for dpkg systems
  53 arch = "i386"
  54 try:
  55     with open(os.path.devnull, "w") as devnull:
  56         dpkg = subprocess.Popen(['dpkg', '--print-architecture'],
  57                                 stderr=devnull, stdout=subprocess.PIPE,
  58                                 universal_newlines=True)
  59
  60         if dpkg.wait() == 0:
  61             arch = dpkg.stdout.read().strip()
  62 except:
  63     pass
  64
  65 ## Create a rootfs
  66 print("Creating rootfs using 'download', arch=%s" % arch)
  67 container.create("download", 0,
  68                  {"dist": "ubuntu",
  69                   "release": "xenial",
  70                   "arch": arch})
  71
  72 assert(container.defined)
  73 assert(container.name == CONTAINER_NAME
  74        == container.get_config_item("lxc.uts.name"))
  75 assert(container.name in lxc.list_containers())
  76
  77 ## Test the config
  78 print("Testing the configuration")
  79 capdrop = container.get_config_item("lxc.cap.drop")
  80 container.clear_config_item("lxc.cap.drop")
  81 container.set_config_item("lxc.cap.drop", capdrop[:-1])
  82 container.append_config_item("lxc.cap.drop", capdrop[-1])
  83 container.save_config()
  84
  85 # A few basic checks of the current state
  86 assert(isinstance(capdrop, list))
  87 assert(capdrop == container.get_config_item("lxc.cap.drop"))
  88
  89 ## Test the networking
  90 print("Testing the networking")
  91
  92 # A few basic checks of the current state
  93 assert("name" in container.get_keys("lxc.net.0"))
  94 assert(len(container.network) == 1)
  95
  96 ## Starting the container
  97 print("Starting the container")
  98 container.start()
  99 container.wait("RUNNING", 3)
 100
 101 # A few basic checks of the current state
 102 assert(container.init_pid > 1)
 103 assert(container.running)
 104 assert(container.state == "RUNNING")
 105
 106
 107 ## Checking IP address
 108 print("Getting the interface names")
 109 assert(set(container.get_interfaces()) == set(('lo', 'eth0')))
 110
 111 ## Checking IP address
 112 print("Getting the IP addresses")
 113
 114 count = 0
 115 ips = []
 116 while not ips or count == 10:
 117     ips = container.get_ips()
 118     time.sleep(1)
 119     count += 1
 120
 121 if os.geteuid():
 122     container.attach_wait(lxc.attach_run_command, ["ifconfig", "eth0"],
 123                           namespaces=(lxc.CLONE_NEWUSER + lxc.CLONE_NEWNET
 124                                       + lxc.CLONE_NEWUTS))
 125 else:
 126     container.attach_wait(lxc.attach_run_command, ["ifconfig", "eth0"],
 127                           namespaces=(lxc.CLONE_NEWNET + lxc.CLONE_NEWUTS))
 128
 129 # A few basic checks of the current state
 130 assert(len(ips) > 0)
 131
 132 ## Test running config
 133 assert(container.name == CONTAINER_NAME
 134        == container.get_config_item("lxc.uts.name")
 135        == container.get_running_config_item("lxc.uts.name"))
 136
 137 ## Testing cgroups a bit
 138 print("Testing cgroup API")
 139 max_mem = container.get_cgroup_item("memory.max_usage_in_bytes")
 140 current_limit = container.get_cgroup_item("memory.limit_in_bytes")
 141 assert(container.set_cgroup_item("memory.limit_in_bytes", max_mem))
 142 assert(container.get_cgroup_item("memory.limit_in_bytes") != current_limit)
 143
 144 ## Freezing the container
 145 print("Freezing the container")
 146 container.freeze()
 147 container.wait("FROZEN", 3)
 148
 149 # A few basic checks of the current state
 150 assert(container.init_pid > 1)
 151 assert(container.running)
 152 assert(container.state == "FROZEN")
 153
 154 ## Unfreezing the container
 155 print("Unfreezing the container")
 156 container.unfreeze()
 157 container.wait("RUNNING", 3)
 158
 159 # A few basic checks of the current state
 160 assert(container.init_pid > 1)
 161 assert(container.running)
 162 assert(container.state == "RUNNING")
 163
 164 if len(sys.argv) > 1 and sys.argv[1] == "--with-console":
 165     ## Attaching to tty1
 166     print("Attaching to tty1")
 167     container.console(tty=1)
 168
 169 ## Shutting down the container
 170 print("Shutting down the container")
 171 if not container.shutdown(3):
 172     container.stop()
 173
 174 if container.running:
 175     print("Stopping the container")
 176     container.stop()
 177     container.wait("STOPPED", 3)
 178
 179 # A few basic checks of the current state
 180 assert(container.init_pid == -1)
 181 assert(not container.running)
 182 assert(container.state == "STOPPED")
 183
 184 ## Snapshotting the container
 185 print("Snapshotting the container")
 186 assert(not container.snapshot_list())
 187 assert(container.snapshot() == "snap0")
 188 assert(len(container.snapshot_list()) == 1)
 189 assert(container.snapshot_restore("snap0") is True)
 190 assert(container.snapshot_destroy("snap0") is True)
 191
 192 ## Cloning the container
 193 print("Cloning the container as '%s'" % CLONE_NAME)
 194 clone = container.clone(CLONE_NAME)
 195 assert(clone is not False)
 196
 197 print ("Renaming the clone to '%s'" % RENAME_NAME)
 198 rename = clone.rename(RENAME_NAME)
 199 rename.start()
 200 rename.stop()
 201 rename.destroy()
 202
 203 ## Destroy the container
 204 print("Destroying the container")
 205 container.destroy()
 206
 207 assert(not container.defined)