test_stream_create.py

#!/usr/bin/env python

# Stdlib
import StringIO
import datetime
import random
import string

# D1
import d1_common.checksum
import d1_common.types.dataoneTypes_v1 as v1
import d1_common.util

# App
import stream_create



PID = 'mytestpid'
SCIOBJ_STR = 'my sciobj bytes'
FORMAT_ID = 'application/octet-stream'
MN_BASE_URL = 'https://192.168.1.128/mn'


def main():
  sysmeta_pyxb = gen_sysmeta()
  mn_client = stream_create.StreamClient(
    base_url=MN_BASE_URL,
  )
  mn_client.createStream(
    PID,
    StringIO.StringIO(SCIOBJ_STR),
    sysmeta_pyxb,
    cert_pub_path='client_cert.pem',
    cert_key_path='client_key_nopassword.pem',
  )


def gen_sysmeta():
  now = datetime.datetime.now()

  sysmeta_pyxb = v1.systemMetadata()
  sysmeta_pyxb.accessPolicy = gen_public_access_policy()
  sysmeta_pyxb.checksum = d1_common.checksum.create_checksum_object(SCIOBJ_STR)
  sysmeta_pyxb.dateSysMetadataModified = now
  sysmeta_pyxb.dateUploaded = now
  sysmeta_pyxb.formatId = FORMAT_ID
  sysmeta_pyxb.identifier = PID
  sysmeta_pyxb.rightsHolder = gen_random_ascii('rights_holder')
  sysmeta_pyxb.size = len(SCIOBJ_STR)
  sysmeta_pyxb.submitter = gen_random_ascii('submitter')

  sysmeta_pyxb.originMemberNode = gen_random_ascii('origin_mn')
  sysmeta_pyxb.authoritativeMemberNode = gen_random_ascii('auth_mn')

  return sysmeta_pyxb


def gen_public_access_policy():
  access_policy_pyxb = v1.accessPolicy()
  access_rule_pyxb = v1.AccessRule()
  access_rule_pyxb.subject.append(d1_common.const.SUBJECT_PUBLIC)
  permission_pyxb = v1.Permission('read')
  access_rule_pyxb.permission.append(permission_pyxb)
  access_policy_pyxb.append(access_rule_pyxb)
  return access_policy_pyxb


def gen_random_ascii(prefix, num_chars=10):
  return '{0}_{1}'.format(
    prefix, ''.join(
      random.
      choice(string.ascii_uppercase + string.ascii_lowercase + string.digits)
      for _ in range(num_chars)
    )
  )


if __name__ == '__main__':
  main()