import os
import sys
import tempfile

from twisted.trial import unittest
from twisted.python import util

from comet.icomet import IHandler
from comet.handler import SpawnCommand

SHELL = '/bin/sh'

class DummyEvent(object):
    def __init__(self, text=None):
        self.text = text or u""

class SpawnCommandProtocolTestCase(unittest.TestCase):
    def test_interface(self):
        self.assertTrue(IHandler.implementedBy(SpawnCommand))

    def test_good_process(self):
        spawn = SpawnCommand(sys.executable)
        d = spawn(DummyEvent())
        d.addCallback(self.assertEqual, True)
        return d

    def test_bad_process(self):
        spawn = SpawnCommand("/not/a/real/executable")
        return self.assertFailure(spawn(DummyEvent()), Exception)

    def test_write_data(self):
        if not os.access(SHELL, os.X_OK):
            raise unittest.SkipTest("Shell not available")
        TEXT = "Test spawn process"
        output_file = tempfile.NamedTemporaryFile()
        def read_data(result):
            try:
                self.assertEqual(output_file.read(), TEXT)
            finally:
                output_file.close()
        spawn = SpawnCommand('/bin/sh', util.sibpath(__file__, "test_spawn.sh"), output_file.name)
        d = spawn(DummyEvent(TEXT))
        d.addCallback(read_data)
        return d

import os
import sys
import tempfile

from twisted.trial import unittest
from twisted.python import util

from comet.icomet import IHandler
from comet.handler import SpawnCommand

SHELL = '/bin/sh'

class DummyEvent(object):
    def __init__(self, text=None):
        self.text = text or u""

class SpawnCommandProtocolTestCase(unittest.TestCase):
    def test_interface(self):
        self.assertTrue(IHandler.implementedBy(SpawnCommand))

    def test_good_process(self):
        spawn = SpawnCommand(sys.executable)
        d = spawn(DummyEvent())
        d.addCallback(self.assertEqual, True)
        return d

    def test_bad_process(self):
        spawn = SpawnCommand("/not/a/real/executable")
        return self.assertFailure(spawn(DummyEvent()), Exception)

    def test_write_data(self):
        if not os.access(SHELL, os.X_OK):
            raise unittest.SkipTest("Shell not available")
        TEXT = u"Test spawn process"
        output_file = tempfile.NamedTemporaryFile()
        def read_data(result):
            try:
                # NamedTemporaryFile is opened in binary mode, so we need to
                # encode the read for comparison.
                self.assertEqual(output_file.read().decode('utf-8'), TEXT)
            finally:
                output_file.close()
        spawn = SpawnCommand('/bin/sh', util.sibpath(__file__, "test_spawn.sh"), output_file.name)
        d = spawn(DummyEvent(TEXT))
        d.addCallback(read_data)
        return d

Convert read data to unicode.
Convert read data to unicode.

The NamedTemporaryFile is in binary mode by default, so the read returns raw
bytes.

  import os
  import sys
  import tempfile
  
  from twisted.trial import unittest
  from twisted.python import util
  
  from comet.icomet import IHandler
  from comet.handler import SpawnCommand
  
  SHELL = '/bin/sh'
  
  class DummyEvent(object):
      def __init__(self, text=None):
          self.text = text or u""
  
  class SpawnCommandProtocolTestCase(unittest.TestCase):
      def test_interface(self):
          self.assertTrue(IHandler.implementedBy(SpawnCommand))
  
      def test_good_process(self):
          spawn = SpawnCommand(sys.executable)
          d = spawn(DummyEvent())
          d.addCallback(self.assertEqual, True)
          return d
  
      def test_bad_process(self):
          spawn = SpawnCommand("/not/a/real/executable")
          return self.assertFailure(spawn(DummyEvent()), Exception)
  
      def test_write_data(self):
          if not os.access(SHELL, os.X_OK):
              raise unittest.SkipTest("Shell not available")
-         TEXT = "Test spawn process"
+         TEXT = u"Test spawn process"
          output_file = tempfile.NamedTemporaryFile()
          def read_data(result):
              try:
+                 # NamedTemporaryFile is opened in binary mode, so we need to
+                 # encode the read for comparison.
-                 self.assertEqual(output_file.read(), TEXT)
+                 self.assertEqual(output_file.read().decode('utf-8'), TEXT)
              finally:
                  output_file.close()
          spawn = SpawnCommand('/bin/sh', util.sibpath(__file__, "test_spawn.sh"), output_file.name)
          d = spawn(DummyEvent(TEXT))
          d.addCallback(read_data)
          return d
  
Convert read data to unicode.
## Code Before:
import os
import sys
import tempfile

from twisted.trial import unittest
from twisted.python import util

from comet.icomet import IHandler
from comet.handler import SpawnCommand

SHELL = '/bin/sh'

class DummyEvent(object):
    def __init__(self, text=None):
        self.text = text or u""

class SpawnCommandProtocolTestCase(unittest.TestCase):
    def test_interface(self):
        self.assertTrue(IHandler.implementedBy(SpawnCommand))

    def test_good_process(self):
        spawn = SpawnCommand(sys.executable)
        d = spawn(DummyEvent())
        d.addCallback(self.assertEqual, True)
        return d

    def test_bad_process(self):
        spawn = SpawnCommand("/not/a/real/executable")
        return self.assertFailure(spawn(DummyEvent()), Exception)

    def test_write_data(self):
        if not os.access(SHELL, os.X_OK):
            raise unittest.SkipTest("Shell not available")
        TEXT = "Test spawn process"
        output_file = tempfile.NamedTemporaryFile()
        def read_data(result):
            try:
                self.assertEqual(output_file.read(), TEXT)
            finally:
                output_file.close()
        spawn = SpawnCommand('/bin/sh', util.sibpath(__file__, "test_spawn.sh"), output_file.name)
        d = spawn(DummyEvent(TEXT))
        d.addCallback(read_data)
        return d

## Instruction:
Convert read data to unicode.
## Code After:
import os
import sys
import tempfile

from twisted.trial import unittest
from twisted.python import util

from comet.icomet import IHandler
from comet.handler import SpawnCommand

SHELL = '/bin/sh'

class DummyEvent(object):
    def __init__(self, text=None):
        self.text = text or u""

class SpawnCommandProtocolTestCase(unittest.TestCase):
    def test_interface(self):
        self.assertTrue(IHandler.implementedBy(SpawnCommand))

    def test_good_process(self):
        spawn = SpawnCommand(sys.executable)
        d = spawn(DummyEvent())
        d.addCallback(self.assertEqual, True)
        return d

    def test_bad_process(self):
        spawn = SpawnCommand("/not/a/real/executable")
        return self.assertFailure(spawn(DummyEvent()), Exception)

    def test_write_data(self):
        if not os.access(SHELL, os.X_OK):
            raise unittest.SkipTest("Shell not available")
        TEXT = u"Test spawn process"
        output_file = tempfile.NamedTemporaryFile()
        def read_data(result):
            try:
                # NamedTemporaryFile is opened in binary mode, so we need to
                # encode the read for comparison.
                self.assertEqual(output_file.read().decode('utf-8'), TEXT)
            finally:
                output_file.close()
        spawn = SpawnCommand('/bin/sh', util.sibpath(__file__, "test_spawn.sh"), output_file.name)
        d = spawn(DummyEvent(TEXT))
        d.addCallback(read_data)
        return d
