<s> from django . conf import settings <EOL> from django . core . exceptions import ImproperlyConfigured <EOL> from django . dispatch import receiver <EOL> from django . template import Context , Template <EOL> from hashlib import md5 <EOL> import glob <EOL> import httplib <EOL> import itertools <EOL> import os <EOL> import re <EOL> import time <EOL> import unittest <EOL> from mediasync import backends , JS_MIMETYPES , listdir_recursive <EOL> from mediasync . backends import BaseClient <EOL> from mediasync . conf import msettings <EOL> from mediasync . signals import pre_sync , post_sync , sass_receiver <EOL> import mediasync <EOL> import mimetypes <EOL> PWD = os . path . abspath ( os . path . dirname ( __file__ ) ) <EOL> EXPIRES_RE = re . compile ( r'<STR_LIT>' ) <EOL> def readfile ( path ) : <EOL> f = open ( path , '<STR_LIT:r>' ) <EOL> content = f . read ( ) <EOL> f . close ( ) <EOL> return content <EOL> class Client ( BaseClient ) : <EOL> def __init__ ( self , * args , ** kwargs ) : <EOL> super ( Client , self ) . __init__ ( * args , ** kwargs ) <EOL> def put ( self , filedata , content_type , remote_path , force = False ) : <EOL> if hasattr ( self , '<STR_LIT>' ) : <EOL> return self . put_callback ( filedata , content_type , remote_path , force ) <EOL> else : <EOL> return True <EOL> def remote_media_url ( self , with_ssl = False ) : <EOL> return ( '<STR_LIT>' if with_ssl else '<STR_LIT:http>' ) + "<STR_LIT>" <EOL> class BackendTestCase ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> msettings [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> def tearDown ( self ) : <EOL> msettings [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> def testInvalidBackend ( self ) : <EOL> self . assertRaises ( ImproperlyConfigured , backends . client ) <EOL> class MockClientTestCase ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> msettings [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> msettings [ '<STR_LIT>' ] = [ ] <EOL> msettings [ '<STR_LIT>' ] = True <EOL> msettings [ '<STR_LIT>' ] = { <EOL> '<STR_LIT>' : ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> '<STR_LIT>' : ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> } <EOL> self . client = backends . client ( ) <EOL> def tearDown ( self ) : <EOL> msettings [ '<STR_LIT>' ] = { } <EOL> def testLocalMediaURL ( self ) : <EOL> self . assertEqual ( self . client . get_local_media_url ( ) , "<STR_LIT>" ) <EOL> def testMediaRoot ( self ) : <EOL> root = getattr ( settings , '<STR_LIT>' , None ) <EOL> if root is None : <EOL> root = getattr ( settings , '<STR_LIT>' , None ) <EOL> self . assertEqual ( self . client . get_media_root ( ) , root ) <EOL> def testMediaURL ( self ) : <EOL> self . assertEqual ( self . client . media_url ( with_ssl = False ) , "<STR_LIT>" ) <EOL> self . assertEqual ( self . client . media_url ( with_ssl = True ) , "<STR_LIT>" ) <EOL> def testSyncableDir ( self ) : <EOL> self . assertFalse ( mediasync . is_syncable_dir ( "<STR_LIT>" ) ) <EOL> self . assertFalse ( mediasync . is_syncable_dir ( "<STR_LIT>" ) ) <EOL> self . assertTrue ( mediasync . is_syncable_dir ( "<STR_LIT:test>" ) ) <EOL> self . assertTrue ( mediasync . is_syncable_dir ( "<STR_LIT>" ) ) <EOL> def testSyncableFile ( self ) : <EOL> self . assertFalse ( mediasync . is_syncable_file ( "<STR_LIT>" ) ) <EOL> self . assertFalse ( mediasync . is_syncable_file ( "<STR_LIT>" ) ) <EOL> self . assertTrue ( mediasync . is_syncable_file ( "<STR_LIT:test>" ) ) <EOL> self . assertTrue ( mediasync . is_syncable_file ( "<STR_LIT>" ) ) <EOL> def testDirectoryListing ( self ) : <EOL> allowed_files = [ <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> ] <EOL> media_dir = os . path . join ( PWD , '<STR_LIT>' ) <EOL> listed_files = list ( mediasync . listdir_recursive ( media_dir ) ) <EOL> self . assertListEqual ( allowed_files , listed_files ) <EOL> def testSync ( self ) : <EOL> to_sync = { <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : msettings [ '<STR_LIT>' ] , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> '<STR_LIT>' : '<STR_LIT>' , <EOL> } <EOL> def generate_callback ( is_forced ) : <EOL> def myput ( filedata , content_type , remote_path , force = is_forced ) : <EOL> self . assertEqual ( content_type , to_sync [ remote_path ] ) <EOL> self . assertEqual ( force , is_forced ) <EOL> if remote_path in msettings [ '<STR_LIT>' ] : <EOL> original = readfile ( os . path . join ( PWD , '<STR_LIT>' , '<STR_LIT>' , remote_path . split ( '<STR_LIT:/>' ) [ <NUM_LIT:1> ] ) ) <EOL> else : <EOL> args = [ PWD , '<STR_LIT>' ] + remote_path . split ( '<STR_LIT:/>' ) <EOL> original = readfile ( os . path . join ( * args ) ) <EOL> self . assertEqual ( filedata , original ) <EOL> return myput <EOL> self . client . put_callback = generate_callback ( is_forced = False ) <EOL> mediasync . sync ( self . client , force = False , verbose = False ) <EOL> self . client . put_callback = generate_callback ( is_forced = True ) <EOL> mediasync . sync ( self . client , force = True , verbose = False ) <EOL> class S3ClientTestCase ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> bucket_hash = md5 ( "<STR_LIT>" % ( int ( time . time ( ) ) , os . environ [ '<STR_LIT>' ] ) ) . hexdigest ( ) <EOL> self . bucket_name = '<STR_LIT>' + bucket_hash <EOL> msettings [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> msettings [ '<STR_LIT>' ] = self . bucket_name <EOL> msettings [ '<STR_LIT>' ] = os . environ [ '<STR_LIT>' ] or None <EOL> msettings [ '<STR_LIT>' ] = os . environ [ '<STR_LIT>' ] or None <EOL> msettings [ '<STR_LIT>' ] = [ ] <EOL> msettings [ '<STR_LIT>' ] = True <EOL> msettings [ '<STR_LIT>' ] = { <EOL> '<STR_LIT>' : ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> '<STR_LIT>' : ( '<STR_LIT>' , '<STR_LIT>' ) , <EOL> } <EOL> self . client = backends . client ( ) <EOL> def testServeRemote ( self ) : <EOL> msettings [ '<STR_LIT>' ] = False <EOL> self . assertEqual ( backends . client ( ) . media_url ( ) , '<STR_LIT>' ) <EOL> msettings [ '<STR_LIT>' ] = True <EOL> self . assertEqual ( backends . client ( ) . media_url ( ) , '<STR_LIT>' % self . bucket_name ) <EOL> def testSync ( self ) : <EOL> cc = "<STR_LIT>" % ( self . client . expiration_days * <NUM_LIT> * <NUM_LIT> ) <EOL> mediasync . sync ( self . client , force = True , verbose = False ) <EOL> self . client . open ( ) <EOL> conn = self . client . get_connection ( ) <EOL> http_conn = httplib . HTTPSConnection ( '<STR_LIT>' ) <EOL> bucket = conn . get_bucket ( self . bucket_name ) <EOL> static_paths = mediasync . listdir_recursive ( os . path . join ( PWD , '<STR_LIT>' ) ) <EOL> joined_paths = msettings [ '<STR_LIT>' ] . iterkeys ( ) <EOL> for path in itertools . chain ( static_paths , joined_paths ) : <EOL> key = bucket . get_key ( path ) <EOL> if path in msettings [ '<STR_LIT>' ] : <EOL> args = [ PWD , '<STR_LIT>' , '<STR_LIT>' , path . split ( '<STR_LIT:/>' ) [ <NUM_LIT:1> ] ] <EOL> else : <EOL> args = [ PWD , '<STR_LIT>' ] + path . split ( '<STR_LIT:/>' ) <EOL> local_content = readfile ( os . path . join ( * args ) ) <EOL> self . assertEqual ( key . read ( ) , local_content ) <EOL> key_meta = key . get_metadata ( '<STR_LIT>' ) or '<STR_LIT>' <EOL> s3_checksum = key_meta . replace ( '<STR_LIT:U+0020>' , '<STR_LIT:+>' ) <EOL> ( hexdigest , b64digest ) = mediasync . checksum ( local_content ) <EOL> self . assertEqual ( s3_checksum , b64digest ) <EOL> http_conn . request ( '<STR_LIT>' , "<STR_LIT>" % ( self . bucket_name , path ) ) <EOL> response = http_conn . getresponse ( ) <EOL> response . read ( ) <EOL> content_type = mimetypes . guess_type ( path ) [ <NUM_LIT:0> ] or msettings [ '<STR_LIT>' ] <EOL> self . assertEqual ( response . getheader ( "<STR_LIT:Content-Type>" , None ) , content_type ) <EOL> expires = response . getheader ( "<STR_LIT>" , None ) <EOL> self . assertRegexpMatches ( expires , EXPIRES_RE ) <EOL> cc_header = response . getheader ( "<STR_LIT>" , None ) <EOL> self . assertEqual ( cc_header , cc ) <EOL> key . delete ( ) <EOL> if content_type in mediasync . TYPES_TO_COMPRESS : <EOL> key = bucket . get_key ( "<STR_LIT>" % path ) <EOL> http_conn . request ( '<STR_LIT>' , "<STR_LIT>" % ( self . bucket_name , path ) ) <EOL> response = http_conn . getresponse ( ) <EOL> response . read ( ) <EOL> key_meta = key . get_metadata ( '<STR_LIT>' ) or '<STR_LIT>' <EOL> s3_checksum = key_meta . replace ( '<STR_LIT:U+0020>' , '<STR_LIT:+>' ) <EOL> self . assertEqual ( s3_checksum , b64digest ) <EOL> key . delete ( ) <EOL> http_conn . close ( ) <EOL> time . sleep ( <NUM_LIT:2> ) <EOL> conn . delete_bucket ( self . bucket_name ) <EOL> self . client . close ( ) <EOL> def testMissingBucket ( self ) : <EOL> del msettings [ '<STR_LIT>' ] <EOL> self . assertRaises ( AssertionError , backends . client ) <EOL> class ProcessorTestCase ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> msettings [ '<STR_LIT>' ] = True <EOL> msettings [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> msettings [ '<STR_LIT>' ] = ( <EOL> '<STR_LIT>' , <EOL> '<STR_LIT>' , <EOL> lambda fd , ct , rp , r : fd . upper ( ) , <EOL> ) <EOL> self . client = backends . client ( ) <EOL> def testJSProcessor ( self ) : <EOL> try : <EOL> import slimmer <EOL> except ImportError : <EOL> self . skipTest ( "<STR_LIT>" ) <EOL> content = """<STR_LIT>""" <EOL> ct = '<STR_LIT>' <EOL> procd = self . client . process ( content , ct , '<STR_LIT>' ) <EOL> self . assertEqual ( procd , '<STR_LIT>' ) <EOL> def testCSSProcessor ( self ) : <EOL> try : <EOL> import slimmer <EOL> except ImportError : <EOL> self . skipTest ( "<STR_LIT>" ) <EOL> content = """<STR_LIT>""" <EOL> ct = '<STR_LIT>' <EOL> procd = self . client . process ( content , ct , '<STR_LIT>' ) <EOL> self . assertEqual ( procd , '<STR_LIT>' ) <EOL> def testCustomProcessor ( self ) : <EOL> procd = self . client . process ( '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . assertEqual ( procd , "<STR_LIT>" ) <EOL> class ClosureCompilerTestCase ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> msettings [ '<STR_LIT>' ] = True <EOL> msettings [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> msettings [ '<STR_LIT>' ] = ( <EOL> '<STR_LIT>' , <EOL> ) <EOL> self . client = backends . client ( ) <EOL> def testCompiler ( self ) : <EOL> content = """<STR_LIT>""" <EOL> for ct in JS_MIMETYPES : <EOL> procd = self . client . process ( content , ct , '<STR_LIT>' ) <EOL> self . assertEqual ( procd , '<STR_LIT>' ) <EOL> def testNotJavascript ( self ) : <EOL> content = """<STR_LIT>""" <EOL> procd = self . client . process ( content , '<STR_LIT>' , '<STR_LIT>' ) <EOL> self . assertEqual ( procd , content ) <EOL> class SignalTestCase ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> msettings [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> self . client = backends . client ( ) <EOL> def tearDown ( self ) : <EOL> root = msettings [ '<STR_LIT>' ] <EOL> for filename in glob . glob ( os . path . join ( root , "<STR_LIT>" ) ) : <EOL> path = filename [ : - <NUM_LIT:4> ] + "<STR_LIT>" <EOL> if os . path . exists ( path ) : <EOL> os . unlink ( path ) <EOL> def testSyncSignals ( self ) : <EOL> self . client . called_presync = False <EOL> self . client . called_postsync = False <EOL> @ receiver ( pre_sync , weak = False ) <EOL> def presync_receiver ( sender , ** kwargs ) : <EOL> self . assertEqual ( self . client , sender ) <EOL> sender . called_presync = True <EOL> @ receiver ( post_sync , weak = False ) <EOL> def postsync_receiver ( sender , ** kwargs ) : <EOL> self . assertEqual ( self . client , sender ) <EOL> sender . called_postsync = True <EOL> mediasync . sync ( self . client , force = True , verbose = False ) <EOL> self . assertTrue ( self . client . called_presync ) <EOL> self . assertTrue ( self . client . called_postsync ) <EOL> def testSassReceiver ( self ) : <EOL> pre_sync . connect ( sass_receiver ) <EOL> mediasync . sync ( self . client , force = True , verbose = False ) <EOL> root = msettings [ '<STR_LIT>' ] <EOL> for sass_path in glob . glob ( os . path . join ( root , "<STR_LIT>" ) ) : <EOL> css_path = sass_path [ : - <NUM_LIT:4> ] + "<STR_LIT>" <EOL> self . assertTrue ( os . path . exists ( css_path ) ) <EOL> class TemplateTagTestCase ( unittest . TestCase ) : <EOL> def setUp ( self ) : <EOL> msettings [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> msettings [ '<STR_LIT>' ] = '<STR_LIT>' <EOL> self . client = backends . client ( ) <EOL> def testMediaURLTag ( self ) : <EOL> pathvar = '<STR_LIT>' <EOL> c = Context ( { '<STR_LIT>' : pathvar } ) <EOL> t = Template ( '<STR_LIT>' ) <EOL> self . assertEqual ( t . render ( c ) , "<STR_LIT>" ) <EOL> t = Template ( '<STR_LIT>' % pathvar ) <EOL> self . assertEqual ( t . render ( c ) , "<STR_LIT>" ) <EOL> t = Template ( '<STR_LIT>' ) <EOL> self . assertEqual ( t . render ( c ) , "<STR_LIT>" ) <EOL> def testCSSTag ( self ) : <EOL> pathvar = '<STR_LIT>' <EOL> c = Context ( { '<STR_LIT>' : pathvar } ) <EOL> t = Template ( '<STR_LIT>' % pathvar ) <EOL> self . assertEqual ( <EOL> t . render ( c ) , <EOL> '<STR_LIT>' % pathvar ) <EOL> t = Template ( '<STR_LIT>' % pathvar ) <EOL> self . assertEqual ( <EOL> t . render ( c ) , <EOL> '<STR_LIT>' % pathvar ) <EOL> t = Template ( '<STR_LIT>' ) <EOL> self . assertEqual ( <EOL> t . render ( c ) , <EOL> '<STR_LIT>' % pathvar ) <EOL> t = Template ( '<STR_LIT>' ) <EOL> self . assertEqual ( <EOL> t . render ( c ) , <EOL> '<STR_LIT>' % pathvar ) <EOL> def testJSTag ( self ) : <EOL> pathvar = '<STR_LIT>' <EOL> c = Context ( { '<STR_LIT>' : pathvar } ) <EOL> t = Template ( '<STR_LIT>' % pathvar ) <EOL> self . assertEqual ( <EOL> t . render ( c ) , <EOL> '<STR_LIT>' % pathvar ) <EOL> t = Template ( '<STR_LIT>' ) <EOL> self . assertEqual ( <EOL> t . render ( c ) , <EOL> '<STR_LIT>' % pathvar ) <EOL> def testMultipleTags ( self ) : <EOL> paths = ( '<STR_LIT>' , '<STR_LIT>' ) <EOL> c = Context ( { '<STR_LIT>' : paths } ) <EOL> t = Template ( '<STR_LIT>' ) <EOL> self . assertEqual ( <EOL> t . render ( c ) , <EOL> '<STR_LIT>' ) </s>
