<s> import os <EOL> import subprocess <EOL> import yaml <EOL> from six import print_ <EOL> from ccmlib import common <EOL> from . import TEST_DIR <EOL> from . import ccmtest <EOL> CLUSTER_PATH = TEST_DIR <EOL> class TestCCMCmd ( ccmtest . Tester ) : <EOL> def __init__ ( self , * args , ** kwargs ) : <EOL> ccmtest . Tester . __init__ ( self , * args , ** kwargs ) <EOL> class TestCCMCreate ( TestCCMCmd ) : <EOL> def tearDown ( self ) : <EOL> p = subprocess . Popen ( [ '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> p . wait ( ) <EOL> def create_cmd ( self , args = None , name = '<STR_LIT:test>' , version = '<STR_LIT>' ) : <EOL> if args is None : <EOL> args = [ ] <EOL> if version : <EOL> args = [ '<STR_LIT>' , '<STR_LIT>' , name , '<STR_LIT>' , version ] + args <EOL> else : <EOL> args = [ '<STR_LIT>' , '<STR_LIT>' , name ] + args <EOL> p = subprocess . Popen ( args , stdout = subprocess . PIPE , stderr = subprocess . PIPE ) <EOL> return p <EOL> def validate_output ( self , process ) : <EOL> stdout , stderr = process . communicate ( ) <EOL> try : <EOL> print_ ( "<STR_LIT>" % stdout ) <EOL> self . assertEqual ( len ( stderr ) , <NUM_LIT:0> ) <EOL> except AssertionError : <EOL> print_ ( "<STR_LIT>" % stderr . strip ( ) ) <EOL> raise <EOL> def cluster_create_version_test ( self ) : <EOL> self . validate_output ( self . create_cmd ( ) ) <EOL> def cluster_create_cassandra_dir_test ( self ) : <EOL> c_dir = common . get_default_path ( ) <EOL> c_dir = os . path . join ( c_dir , '<STR_LIT>' ) <EOL> c_dir = os . path . join ( c_dir , os . listdir ( c_dir ) [ <NUM_LIT:0> ] ) <EOL> args = [ '<STR_LIT>' , c_dir ] <EOL> self . validate_output ( self . create_cmd ( args , version = None ) ) <EOL> def cluster_create_populate_test ( self ) : <EOL> args = [ '<STR_LIT>' , '<STR_LIT:3>' ] <EOL> self . validate_output ( self . create_cmd ( args ) ) <EOL> def cluster_create_no_switch_test ( self ) : <EOL> self . create_cmd ( args = None , name = '<STR_LIT>' ) <EOL> args = [ '<STR_LIT>' ] <EOL> self . validate_output ( self . create_cmd ( args ) ) <EOL> self . assertEqual ( '<STR_LIT>' , common . current_cluster_name ( common . get_default_path ( ) ) ) <EOL> p = subprocess . Popen ( [ '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> p . wait ( ) <EOL> p = subprocess . Popen ( [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT:test>' ] ) <EOL> p . wait ( ) <EOL> p = subprocess . Popen ( [ '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> p . wait ( ) <EOL> def cluster_create_start_test ( self ) : <EOL> args = [ '<STR_LIT>' , '<STR_LIT:1>' , '<STR_LIT>' ] <EOL> self . validate_output ( self . create_cmd ( args ) ) <EOL> pidfile = os . path . join ( common . get_default_path ( ) , '<STR_LIT:test>' , '<STR_LIT>' , '<STR_LIT>' ) <EOL> with open ( pidfile , '<STR_LIT:r>' ) as f : <EOL> pid = int ( f . readline ( ) . strip ( ) ) <EOL> os . kill ( pid , <NUM_LIT:0> ) <EOL> def cluster_create_debug_start_test ( self ) : <EOL> args = [ '<STR_LIT>' , '<STR_LIT:1>' , '<STR_LIT>' , '<STR_LIT>' ] <EOL> p = self . create_cmd ( args ) <EOL> stdout , stderr = p . communicate ( ) <EOL> print_ ( "<STR_LIT>" % stdout ) <EOL> self . assertGreater ( len ( stdout ) , <NUM_LIT> ) <EOL> self . assertEqual ( len ( stderr ) , <NUM_LIT:0> ) <EOL> def cluster_create_vnodes_test ( self ) : <EOL> args = [ '<STR_LIT>' , '<STR_LIT:1>' , '<STR_LIT>' ] <EOL> self . validate_output ( self . create_cmd ( args ) ) <EOL> yaml_path = os . path . join ( common . get_default_path ( ) , '<STR_LIT:test>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ) <EOL> with open ( yaml_path , '<STR_LIT:r>' ) as f : <EOL> data = yaml . load ( f ) <EOL> self . assertEqual ( <NUM_LIT> , data [ '<STR_LIT>' ] ) </s>
