<commit_before>import csv
import os
import subprocess

# Gather the packages to test.

PREFIX = './packages/node_modules/'
CISCOSPARK = os.path.join(PREFIX, '@ciscospark')
WEBEX = os.path.join(PREFIX, '@webex')

PROD_ENV_VARS = {
  # 'ACL_SERVICE_URL': 'https://acl-a.wbx2.com/acl/api/v1', ?
  'ATLAS_SERVICE_URL': 'https://atlas-a.wbx2.com/admin/api/v1',
  'CONVERSATION_SERVICE': 'https://conv-a.wbx2.com/conversation/api/v1',
  'ENCRYPTION_SERVICE_URL': 'https://encryption-a.wbx2.com',
  'IDBROKER_BASE_URL': 'https://idbroker.webex.com',
  'IDENTITY_BASE_URL': 'https://identity.webex.com',
  'WDM_SERVICE_URL': 'https://wdm-a.wbx2.com/wdm/api/v1',
  # Logging
  'ENABLE_VERBOSE_NETWORK_LOGGING': 'true'
}

INT_ENV_VARS = {
  'ACL_SERVICE_URL': 'https://acl-intb.ciscospark.com/acl/api/v1',
  'ATLAS_SERVICE_URL': 'https://atlas-intb.ciscospark.com/admin/api/v1',
  'CONVERSATION_SERVICE': 'https://conversation-intb.ciscospark.com/conversation/api/v1',
  'ENCRYPTION_SERVICE_URL': 'https://encryption-intb.ciscospark.com/encryption/api/v1',
  'IDBROKER_BASE_URL': 'https://idbrokerbts.webex.com',
  'IDENTITY_BASE_URL': 'https://identitybts.webex.com',
  'WDM_SERVICE_URL': 'https://wdm-intb.ciscospark.com/wdm/api/v1',
  # Logging
  'ENABLE_VERBOSE_NETWORK_LOGGING': 'true'
}

OUTPUT_DIR = 'output'
OUTPUT_FILE_PATH = os.path.join(OUTPUT_DIR, 'test-comparison.csv')

# TEST_COMMAND = 'npm test -- --package %s'
TEST_COMMAND = 'npm test -- --package %s --node'

def get_package_names(path_name):
  namespace = path_name.replace(PREFIX, '')
  return [os.path.join(namespace, name) for name in os.listdir(path_name) if os.path.isdir(os.path.join(path_name, name))]

def run_subprocess(bash_command, env_vars):
  env = os.environ.copy()
  env.update(env_vars)
  process = subprocess.Popen(bash_command.split(), stdout=subprocess.PIPE, env=env)

  output, error = process.communicate()
  return process.returncode # , output, error

class bcolors:
    HEADER = '\033[95m'
    OKBLUE = '\033[94m'
    OKGREEN = '\033[92m'
    WARNING = '\033[93m'
    FAIL = '\033[91m'
    ENDC = '\033[0m'
    BOLD = '\033[1m'
    UNDERLINE = '\033[4m'

def print_result(return_code, prefix='Tests are a...'):
  if return_code == 0:
    print(bcolors.OKGREEN + prefix + 'success.' + bcolors.ENDC)
  else:
    print(bcolors.FAIL + prefix + 'failure.' + bcolors.ENDC)

def main():
  ciscospark_packages = get_package_names(CISCOSPARK)
  webex_packages = get_package_names(WEBEX)
  packages = ciscospark_packages + webex_packages
  print('Testing %d packages...' % len(packages))

  try:
    os.mkdir(OUTPUT_DIR)
  except OSError:
    pass


  with open(OUTPUT_FILE_PATH, 'wb') as csvfile:
    writer = csv.writer(csvfile, quoting=csv.QUOTE_MINIMAL)
    writer.writerow(['Package', 'Production exit code', 'Integration exit code'])
    for package in packages:
      bash_command = TEST_COMMAND % package

      # Test production.
      print(bcolors.OKBLUE + 'Testing `%s` on production...' % package + bcolors.ENDC)
      prod_return_code = run_subprocess(bash_command, PROD_ENV_VARS)
      print_result(prod_return_code, prefix='Testing `%s` on production...' % package)

      # Test integration.
      print(bcolors.OKBLUE + 'Testing `%s` on integration...' % package + bcolors.ENDC)
      int_return_code = run_subprocess(bash_command, INT_ENV_VARS)
      print_result(int_return_code, prefix='Testing `%s` on integration...' % package)

      writer.writerow([package, prod_return_code, int_return_code])
      csvfile.flush()

  print('Wrote output to: %s' % OUTPUT_FILE_PATH)
  print('Done.')

if __name__ == "__main__":
  main()<commit_msg>refactor(test.py): create run_test function for all environments<commit_after>import csv
import os
import subprocess
import threading

# Gather the packages to test.

PREFIX = './packages/node_modules/'
CISCOSPARK = os.path.join(PREFIX, '@ciscospark')
WEBEX = os.path.join(PREFIX, '@webex')

PROD_ENV_VARS = {
  # 'ACL_SERVICE_URL': 'https://acl-a.wbx2.com/acl/api/v1', ?
  'ATLAS_SERVICE_URL': 'https://atlas-a.wbx2.com/admin/api/v1',
  'CONVERSATION_SERVICE': 'https://conv-a.wbx2.com/conversation/api/v1',
  'ENCRYPTION_SERVICE_URL': 'https://encryption-a.wbx2.com',
  'IDBROKER_BASE_URL': 'https://idbroker.webex.com',
  'IDENTITY_BASE_URL': 'https://identity.webex.com',
  'WDM_SERVICE_URL': 'https://wdm-a.wbx2.com/wdm/api/v1',
  # Logging
  'ENABLE_VERBOSE_NETWORK_LOGGING': 'true'
}

INT_ENV_VARS = {
  'ACL_SERVICE_URL': 'https://acl-intb.ciscospark.com/acl/api/v1',
  'ATLAS_SERVICE_URL': 'https://atlas-intb.ciscospark.com/admin/api/v1',
  'CONVERSATION_SERVICE': 'https://conversation-intb.ciscospark.com/conversation/api/v1',
  'ENCRYPTION_SERVICE_URL': 'https://encryption-intb.ciscospark.com/encryption/api/v1',
  'IDBROKER_BASE_URL': 'https://idbrokerbts.webex.com',
  'IDENTITY_BASE_URL': 'https://identitybts.webex.com',
  'WDM_SERVICE_URL': 'https://wdm-intb.ciscospark.com/wdm/api/v1',
  # Logging
  'ENABLE_VERBOSE_NETWORK_LOGGING': 'true'
}

OUTPUT_DIR = 'output'
OUTPUT_FILE_PATH = os.path.join(OUTPUT_DIR, 'test-comparison.csv')

# TEST_COMMAND = 'npm test -- --package %s'
TEST_COMMAND = 'npm test -- --package %s --node'

def get_package_names(path_name):
  namespace = path_name.replace(PREFIX, '')
  return [os.path.join(namespace, name) for name in os.listdir(path_name) if os.path.isdir(os.path.join(path_name, name))]

def run_subprocess(bash_command, env_vars):
  env = os.environ.copy()
  env.update(env_vars)
  process = subprocess.Popen(bash_command.split(), stdout=subprocess.PIPE, env=env)

  output, error = process.communicate()
  return process.returncode # , output, error

class bcolors:
    HEADER = '\033[95m'
    OKBLUE = '\033[94m'
    OKGREEN = '\033[92m'
    WARNING = '\033[93m'
    FAIL = '\033[91m'
    ENDC = '\033[0m'
    BOLD = '\033[1m'
    UNDERLINE = '\033[4m'

def print_result(return_code, prefix='Tests are a...'):
  if return_code == 0:
    print(bcolors.OKGREEN + prefix + 'success.' + bcolors.ENDC)
  else:
    print(bcolors.FAIL + prefix + 'failure.' + bcolors.ENDC)

def run_test(package, environment):
  env_vars = INT_ENV_VARS if environment is 'integration' else PROD_ENV_VARS
  print(bcolors.OKBLUE + 'Testing `%s` on %s...' % (package, environment) + bcolors.ENDC)
  bash_command = TEST_COMMAND % package
  return_code = run_subprocess(bash_command, env_vars)
  print_result(return_code, prefix='Testing `%s` on %s...' % (package, environment))
  return return_code

def main():
  ciscospark_packages = get_package_names(CISCOSPARK)
  webex_packages = get_package_names(WEBEX)
  packages = ciscospark_packages + webex_packages
  print('Testing %d packages...' % len(packages))

  try:
    os.mkdir(OUTPUT_DIR)
  except OSError:
    pass

  threads = []

  with open(OUTPUT_FILE_PATH, 'wb') as csvfile:
    writer = csv.writer(csvfile, quoting=csv.QUOTE_MINIMAL)
    writer.writerow(['Package', 'Production exit code', 'Integration exit code'])
    for package in packages:
      prod_return_code = run_test(package, 'production')
      int_return_code = run_test(package, 'integration')
      writer.writerow([package, prod_return_code, int_return_code])
      csvfile.flush()

  print('Wrote output to: %s' % OUTPUT_FILE_PATH)
  print('Done.')

if __name__ == "__main__":
  main()
{'fraction_non_alphanumeric': 0.15433200604146644, 'fraction_numerical': 0.014279829740491556, 'mean_word_length': 4.411589895988113, 'pattern_counts': {'":': 2, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 28, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1227855', 'n_tokens_mistral': 2864, 'n_tokens_neox': 2773, 'n_words': 463}