recipe_modules/json/examples/full.py

# Copyright 2014 The LUCI Authors. All rights reserved.
# Use of this source code is governed under the Apache License, Version 2.0
# that can be found in the LICENSE file.

DEPS = [
  'json',
  'path',
  'properties',
  'python',
  'raw_io',
  'step',
]

FULLWIDTH_Z = '\xef\xbb\xbf\xef\xbd\x9a'

def RunSteps(api):
  step_result = api.step('echo1', ['echo', '[1, 2, 3]'],
      stdout=api.json.output())
  assert step_result.stdout == [1, 2, 3], step_result.stdout

  # have to provide a default value for example_dumps because the
  # 'example_tests' unittest runs all these example recipes without any input
  # properties.
  api.step('echo_dumps_property', [
    'echo', api.properties.get('example_dumps', '[100]')])

  # Example demonstrating the usage of step_test_data for json stdout.
  step_result = api.step('echo2', ['echo', '[2, 3, 4]'],
      step_test_data=lambda: api.json.test_api.output_stream([2, 3, 4]),
      stdout=api.json.output())
  assert step_result.stdout == [2, 3, 4]

  assert api.json.is_serializable('foo')
  assert not api.json.is_serializable(set(['foo', 'bar', 'baz']))

  # Example demonstrating multiple named json output files.
  result = api.python.inline(
      'foo',
      """
      import json
      import sys
      with open(sys.argv[1], 'w') as f:
        f.write(json.dumps([1, 2, 3]))
      with open(sys.argv[2], 'w') as f:
        f.write(json.dumps(['x', 'y', %s]))
      """ % repr(FULLWIDTH_Z),
      args=[api.json.output(name='1'), api.json.output(name='2')],
  )
  assert result.json.outputs['1'] == [1, 2, 3]
  assert result.json.outputs['2'] == ['x', 'y', FULLWIDTH_Z]
  assert not hasattr(result.json, 'output')

  example_dict = {'x': 1, 'y': 2}

  # json.input(json_data) expands to a path containing that rendered json
  step_result = api.step('json through',
    ['cat', api.json.input(example_dict)],
    stdout=api.json.output(),
    step_test_data=lambda: api.json.test_api.output_stream(example_dict))
  assert step_result.stdout == example_dict

  # json.read reads a file containing json data.
  leak_path = api.path['tmp_base'].join('temp.json')
  api.step('write json to file',
    ['cat', api.json.input(example_dict)],
    stdout=api.raw_io.output(leak_to=leak_path))
  step_result = api.json.read(
      'read json from file we just wrote', leak_path,
      step_test_data=lambda: api.json.test_api.output(example_dict))
  assert step_result.json.output == example_dict

  # can leak directly to a file
  step_result = api.step('leaking json', [
    'python', api.resource('cool_script.py'),
    '{"x":1,"y":2}',
    api.json.output(leak_to=api.path['tmp_base'].join('leak.json')),
  ])
  assert step_result.json.output == example_dict

  # invalid data gets rendered
  step_result = api.step('invalid json', [
    'python', api.resource('cool_script.py'),
    '{"here is some total\ngarbage',
    api.json.output(),
  ])
  assert step_result.json.output is None

  step_result = api.step(
    'backing file missing',
    [
      'python', api.resource('cool_script.py'),
      'file missing',
      api.json.output(leak_to='/this/file/doesnt/exist'),
    ],
    ok_ret=(1,))
  assert step_result.json.output is None


def GenTests(api):
  yield (
    api.test('basic')
    + api.properties(
      example_dumps=api.json.dumps({
        'hello': 'world',
        'cool': [1, 2, 3],
      })
    )
    + api.step_data('echo1', stdout=api.json.output([1, 2, 3]))
    + api.step_data(
      'foo',
      api.json.output([1, 2, 3], name='1') +
      api.json.output(['x', 'y', FULLWIDTH_Z], name='2'),
    )
    + api.step_data(
      'leaking json',
      api.json.output({'x': 1, 'y': 2}),
    )
    + api.step_data(
      'invalid json',
      api.json.invalid('{"here is some total\ngarbage'),
    )
    + api.step_data(
      'backing file missing',
      api.json.backing_file_missing(),
      retcode=1,
    )
  )