text: # Licensed to the Apache Software Foundation (ASF) under one
  # or more contributor license agreements.  See the NOTICE file
  # distributed with this work for additional information
  # regarding copyright ownership.  The ASF licenses this file
  # to you under the Apache License, Version 2.0 (the
  # "License"); you may not use this file except in compliance
  # with the License.  You may obtain a copy of the License at
  #
  # https://www.apache.org/licenses/LICENSE-2.0
  #
  # Unless required by applicable law or agreed to in writing, software
  # distributed under the License is distributed on an "AS IS" BASIS,
  # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  # See the License for the specific language governing permissions and
  # limitations under the License.
  
  
  
  import os
  import subprocess
  import sys
  import time
  import unittest
  
  import set_avro_test_path
  
  
  class TestTetherTask(unittest.TestCase):
    """
    TODO: We should validate the the server response by looking at stdout
    """
    def test1(self):
      """
      Test that the thether_task is working. We run the mock_tether_parent in a separate
      subprocess
      """
      from avro import tether
      from avro import io as avio
      from avro import schema
      from avro.tether import HTTPRequestor,inputProtocol, find_port
  
      import StringIO
      import mock_tether_parent
      from word_count_task import WordCountTask
  
      task=WordCountTask()
  
      proc=None
      try:
        # launch the server in a separate process
        # env["AVRO_TETHER_OUTPUT_PORT"]=output_port
        env=dict()
        env["PYTHONPATH"]=':'.join(sys.path)
        server_port=find_port()
  
        pyfile=mock_tether_parent.__file__
        proc=subprocess.Popen(["python", pyfile,"start_server","{0}".format(server_port)])
        input_port=find_port()
  
        print "Mock server started process pid={0}".format(proc.pid)
        # Possible race condition? open tries to connect to the subprocess before the subprocess is fully started
        # so we give the subprocess time to start up
        time.sleep(1)
        task.open(input_port,clientPort=server_port)
  
        # TODO: We should validate that open worked by grabbing the STDOUT of the subproces
        # and ensuring that it outputted the correct message.
  
        #***************************************************************
        # Test the mapper
        task.configure(tether.TaskType.MAP,str(task.inschema),str(task.midschema))
  
        # Serialize some data so we can send it to the input function
        datum="This is a line of text"
        writer = StringIO.StringIO()
        encoder = avio.BinaryEncoder(writer)
        datum_writer = avio.DatumWriter(task.inschema)
        datum_writer.write(datum, encoder)
  
        writer.seek(0)
        data=writer.read()
  
        # Call input to simulate calling map
        task.input(data,1)
  
        # Test the reducer
        task.configure(tether.TaskType.REDUCE,str(task.midschema),str(task.outschema))
  
        # Serialize some data so we can send it to the input function
        datum={"key":"word","value":2}
        writer = StringIO.StringIO()
        encoder = avio.BinaryEncoder(writer)
        datum_writer = avio.DatumWriter(task.midschema)
        datum_writer.write(datum, encoder)
  
        writer.seek(0)
        data=writer.read()
  
        # Call input to simulate calling reduce
        task.input(data,1)
  
        task.complete()
  
        # try a status
        task.status("Status message")
  
      except Exception as e:
        raise
      finally:
        # close the process
        if not(proc is None):
          proc.kill()
  
        pass
  
  if __name__ == '__main__':
    unittest.main()
  