Testea login y subida de archivo a Object Storage

openstack_object_storage_test.py

"""
Tests login and upload to an OpenStack Object Storage instance.
"""
import random
import subprocess
import unittest

import cloudfiles

from cloudfiles.errors import NoSuchContainer, AuthenticationFailed

USERNAME = "system:root"
PASSWORD = "testpass"
AUTH_URL =  "https://192.168.122.233:11000/v1.0"

def login():
    """
    Authenticates with USERNAME and PASSWORD.
    """
    return cloudfiles.get_connection(USERNAME, PASSWORD, authurl=AUTH_URL)

class BaseTestCase(unittest.TestCase):
    """Base class for tests."""

class ObjectStorageLoginTest(BaseTestCase):
    """Test valid and invalid login"""
    
    def test_login(self): # pylint: disable=R0201
        """Tests that login works."""
        login()
    
    def test_wrong_login(self):
        """
        Tests that that login doesn't works with invalid username and password.
        """
        self.assertRaises(AuthenticationFailed, cloudfiles.get_connection, ("invalid_name",
            "invalid_password"), {'authurl': AUTH_URL, })

class ObjectStorageContainerAndUploadTest(BaseTestCase):
    """Test upload and simple operations with containers."""
    
    def test_create_container(self):
        """Test upload and simple operations with containers."""
        conn = login()
        containers = conn.get_all_containers()
        
        # Creates a new container, with a random name, after checking that
        # doesn't exists a container with that name.
        new_container_name = "container_%d" % random.randint(0, 999999999)
        self.assertRaises(NoSuchContainer, conn.get_container, [new_container_name])
        
        conn.create_container(new_container_name)
        new_container = conn.get_container(new_container_name)
        
        self.assertTrue(len(containers)+1, len(conn.get_all_containers()))
        
        # Now upload a new file...
        object_name = "some_file_%d.raw" % random.randint(0, 999999999)
        created_object = new_container.create_object(object_name)
        created_object.content_type = "application/octet-stream"
        pipe = subprocess.Popen("dd if=/dev/urandom bs=1024k count=1 2> /dev/null",
            shell=True, stdout=subprocess.PIPE).stdout
        created_object.send(pipe)
        
        # ...and check that the file exists
        obj = new_container.get_object(object_name)
        self.assertEquals(obj.name, object_name)
        self.assertEquals(obj.size, 1024*1024)

if __name__ == '__main__':
    unittest.main()