mojo/python/tests/bindings_interface_unittest.py

# Copyright 2014 The Chromium Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.

import gc
import weakref

# pylint: disable=F0401,E0611
import mojo_bindings.promise as promise
import mojo_system as system
import mojo_unittest
import regression_tests_mojom
import sample_factory_mojom
import sample_service_mojom


def _BuildProxy(impl):
  pipe = system.MessagePipe()
  impl.__class__.manager.Bind(impl, pipe.handle0)
  return impl.__class__.manager.Proxy(pipe.handle1)


def _ExtractValue(p):
  container = []
  @promise.async
  def GetValue(value):
    container.append(value)
  GetValue(p)
  assert len(container)
  return container[0]


class EmptyServiceImpl(sample_service_mojom.Service):

  def __init__(self):
    pass


class ServiceImpl(sample_service_mojom.Service):

  def __init__(self):
    pass

  # pylint: disable=C0102,W0613
  def Frobinate(self, foo, baz, port):
    return baz


class NamedObjectImpl(sample_factory_mojom.NamedObject):

  def __init__(self):
    self.name = 'name'

  def SetName(self, name):
    self.name = name

  def GetName(self):
    return self.name


class DelegatingNamedObject(sample_factory_mojom.NamedObject):

  def __init__(self):
    self.proxy = _BuildProxy(NamedObjectImpl())

  def SetName(self, name):
    self.proxy.SetName(name)

  def GetName(self):
    return self.proxy.GetName()

class InterfaceTest(mojo_unittest.MojoTestCase):

  def testBaseInterface(self):
    service = sample_service_mojom.Service()
    with self.assertRaises(AttributeError):
      service.NotExisting()
    with self.assertRaises(NotImplementedError):
      service.Frobinate()

  def testEmpty(self):
    service = EmptyServiceImpl()
    with self.assertRaises(NotImplementedError):
      service.Frobinate()

  def testServiceWithReturnValue(self):
    proxy = _BuildProxy(DelegatingNamedObject())
    p1 = proxy.GetName()

    self.assertEquals(p1.state, promise.Promise.STATE_PENDING)
    self.loop.RunUntilIdle()
    self.assertEquals(p1.state, promise.Promise.STATE_FULLFILLED)
    name = _ExtractValue(p1)
    self.assertEquals(name, 'name')

    proxy.SetName('hello')
    p2 = proxy.GetName()

    self.assertEquals(p2.state, promise.Promise.STATE_PENDING)
    self.loop.RunUntilIdle()
    self.assertEquals(p2.state, promise.Promise.STATE_FULLFILLED)
    name = _ExtractValue(p2)
    self.assertEquals(name, 'hello')

  def testCloseProxy(self):
    named_object_impl = NamedObjectImpl()
    proxy = _BuildProxy(named_object_impl)
    response = proxy.GetName()
    proxy.manager.Close()

    self.assertEquals(response.state, promise.Promise.STATE_REJECTED)

  def testCloseImplementationWithResponse(self):
    impl = DelegatingNamedObject()
    proxy = _BuildProxy(impl)
    p1 = proxy.GetName()

    self.assertEquals(p1.state, promise.Promise.STATE_PENDING)

    impl.manager.Close()
    self.loop.RunUntilIdle()

    self.assertEquals(p1.state, promise.Promise.STATE_REJECTED)